benevoles/backend/tests/test_import_service.py

from sqlmodel import select

from app.importing.parser import ParsedTable
from app.importing.schemas import ColumnMapping, ImportMapping
from app.importing.service import run_import, suggest_mapping, validate_mapping
from app.models import Volunteer

HEADERS = ["Prénom", "Nom", "Email", "Téléphone", "Remarque"]


def _mapping(dry_run: bool = False) -> ImportMapping:
    return ImportMapping(
        dry_run=dry_run,
        columns=ColumnMapping(
            first_name="Prénom",
            last_name="Nom",
            email="Email",
            phone="Téléphone",
            comment="Remarque",
        ),
    )


def _table(rows: list[list[str]]) -> ParsedTable:
    return ParsedTable(headers=HEADERS, rows=rows)


def _volunteers(session, edition):
    return list(
        session.exec(
            select(Volunteer).where(Volunteer.edition_id == edition.id)
        )
    )


def test_suggest_mapping_matches_common_headers():
    suggestion = suggest_mapping(
        ["Prénom", "Nom de famille", "Adresse e-mail", "Numéro de portable"]
    )
    assert suggestion["first_name"] == "Prénom"
    assert suggestion["last_name"] == "Nom de famille"
    assert suggestion["email"] == "Adresse e-mail"
    assert suggestion["phone"] == "Numéro de portable"


def test_happy_path_creates_volunteers(session, edition):
    report = run_import(
        session=session,
        edition=edition,
        table=_table(
            [
                ["Alex", "Martin", "alex@ex.fr", "0600000000", "arrive tôt"],
                ["Sam", "Nguyen", "sam@ex.fr", "", ""],
            ]
        ),
        mapping=_mapping(),
    )

    assert (report.created, report.updated, report.skipped) == (2, 0, 0)
    volunteers = {v.email: v for v in _volunteers(session, edition)}
    assert set(volunteers) == {"alex@ex.fr", "sam@ex.fr"}
    assert volunteers["alex@ex.fr"].phone == "0600000000"
    assert volunteers["alex@ex.fr"].comment == "arrive tôt"


def test_dry_run_persists_nothing(session, edition):
    report = run_import(
        session=session,
        edition=edition,
        table=_table([["Alex", "Martin", "alex@ex.fr", "", ""]]),
        mapping=_mapping(dry_run=True),
    )
    assert report.created == 1
    assert report.dry_run is True
    assert _volunteers(session, edition) == []


def test_existing_email_is_updated_not_duplicated(session, edition):
    session.add(
        Volunteer(
            edition_id=edition.id,
            email="alex@ex.fr",
            first_name="Alexandre",
            last_name="M.",
            phone="0600000000",
        )
    )
    session.commit()

    report = run_import(
        session=session,
        edition=edition,
        table=_table([["Alex", "Martin", "ALEX@ex.fr", "", ""]]),
        mapping=_mapping(),
    )

    assert (report.created, report.updated) == (0, 1)
    assert report.rows[0].issues[0].code == "import.row.updated_existing"
    volunteers = _volunteers(session, edition)
    assert len(volunteers) == 1
    assert volunteers[0].first_name == "Alex"
    assert volunteers[0].last_name == "Martin"


def test_duplicate_email_within_file_is_skipped(session, edition):
    report = run_import(
        session=session,
        edition=edition,
        table=_table(
            [
                ["Alex", "Martin", "dup@ex.fr", "", ""],
                ["Alex", "Doublon", "dup@ex.fr", "", ""],
            ]
        ),
        mapping=_mapping(),
    )
    assert (report.created, report.skipped) == (1, 1)
    skipped = report.rows[1]
    assert skipped.action == "SKIPPED"
    assert skipped.issues[0].code == "import.row.duplicate_email_in_file"
    assert skipped.issues[0].params["first_line"] == 2


def test_row_without_email_matching_existing_name_warns_but_creates(
    session, edition
):
    session.add(
        Volunteer(
            edition_id=edition.id,
            first_name="Jean",
            last_name="Dupont",
        )
    )
    session.commit()

    report = run_import(
        session=session,
        edition=edition,
        table=_table([["jean", "DUPONT", "", "", ""]]),
        mapping=_mapping(),
    )
    assert report.created == 1
    assert report.rows[0].issues[0].code == "import.row.possible_duplicate_name"
    assert len(_volunteers(session, edition)) == 2


def test_missing_required_field_is_skipped(session, edition):
    report = run_import(
        session=session,
        edition=edition,
        table=_table([["", "Martin", "x@ex.fr", "", ""]]),
        mapping=_mapping(),
    )
    assert report.skipped == 1
    assert report.rows[0].issues[0].code == "import.row.missing_required_field"
    assert _volunteers(session, edition) == []


def test_invalid_email_is_skipped(session, edition):
    report = run_import(
        session=session,
        edition=edition,
        table=_table([["Alex", "Martin", "pas-un-email", "", ""]]),
        mapping=_mapping(),
    )
    assert report.skipped == 1
    assert report.rows[0].issues[0].code == "import.row.invalid_email"


def test_validate_mapping_reports_missing_and_unknown_columns():
    mapping = ImportMapping(
        columns=ColumnMapping(first_name="Prénom", last_name="", email="Absente")
    )
    errors = validate_mapping(_table([]), mapping)
    codes = {e.code for e in errors}
    assert "import.mapping.required_field_missing" in codes
    assert "import.mapping.column_not_found" in codes