benevoles/backend/tests/test_import_service.py

import datetime as dt

from sqlmodel import select

from app.importing.parser import ParsedTable
from app.importing.schemas import (
    AvailabilityColumnMapping,
    ColumnMapping,
    ImportMapping,
)
from app.importing.service import run_import, suggest_mapping, validate_mapping
from app.models import Availability, Skill, Volunteer

HEADERS = ["Prénom", "Nom", "Email", "Taille", "Compétences", "Ven matin", "Ven aprem"]


def _mapping(dry_run: bool = False) -> ImportMapping:
    return ImportMapping(
        dry_run=dry_run,
        columns=ColumnMapping(
            first_name="Prénom",
            last_name="Nom",
            email="Email",
            tshirt_size="Taille",
            skills="Compétences",
        ),
        availability_columns=[
            AvailabilityColumnMapping(
                header="Ven matin", date=dt.date(2026, 7, 10),
                day_part_code="MORNING",
            ),
            AvailabilityColumnMapping(
                header="Ven aprem", date=dt.date(2026, 7, 10),
                day_part_code="AFTERNOON",
            ),
        ],
    )


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_skills_and_availabilities(session, edition):
    report = run_import(
        session=session,
        edition=edition,
        table=_table(
            [
                ["Alex", "Martin", "alex@ex.fr", "M", "SST, Permis B", "x", "non"],
                ["Sam", "Nguyen", "sam@ex.fr", "L", "SST", "si besoin", ""],
            ]
        ),
        mapping=_mapping(),
    )

    assert (report.created, report.updated, report.skipped) == (2, 0, 0)
    assert report.skills_created == ["PERMIS_B", "SST"]
    assert report.availabilities_written == 3  # x, non, si besoin

    volunteers = {v.email: v for v in _volunteers(session, edition)}
    assert set(volunteers) == {"alex@ex.fr", "sam@ex.fr"}
    assert volunteers["alex@ex.fr"].tshirt_size.value == "M"
    assert {s.code for s in volunteers["alex@ex.fr"].skills} == {"SST", "PERMIS_B"}

    avails = session.exec(
        select(Availability).where(
            Availability.volunteer_id == volunteers["alex@ex.fr"].id
        )
    ).all()
    assert {a.status.value for a in avails} == {"AVAILABLE", "UNAVAILABLE"}


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


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", "M", "SST", "x", ""]]),
        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", "M", "", "x", ""],
                ["Alex", "Doublon", "dup@ex.fr", "L", "", "x", ""],
            ]
        ),
        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", "", "", "", "x", ""]]),
        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", "", "", "x", ""]]),
        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", "", "", "x", ""]]),
        mapping=_mapping(),
    )
    assert report.skipped == 1
    assert report.rows[0].issues[0].code == "import.row.invalid_email"


def test_unknown_tshirt_size_warns_and_still_imports(session, edition):
    report = run_import(
        session=session,
        edition=edition,
        table=_table([["Alex", "Martin", "a@ex.fr", "Large", "", "x", ""]]),
        mapping=_mapping(),
    )
    assert report.created == 1
    assert report.rows[0].issues[0].code == "import.row.invalid_tshirt_size"
    assert _volunteers(session, edition)[0].tshirt_size is None


def test_unknown_availability_value_is_flagged(session, edition):
    report = run_import(
        session=session,
        edition=edition,
        table=_table([["Alex", "Martin", "a@ex.fr", "", "", "bof", "x"]]),
        mapping=_mapping(),
    )
    codes = [i.code for i in report.rows[0].issues]
    assert "import.row.unknown_availability_value" in codes
    assert report.availabilities_written == 1  # seule "Ven aprem" a été retenue


def test_validate_mapping_reports_bad_day_part_and_date(session, edition):
    from app.models import DayPart

    day_parts = list(
        session.exec(select(DayPart).where(DayPart.edition_id == edition.id))
    )
    mapping = ImportMapping(
        columns=ColumnMapping(first_name="Prénom", last_name="Nom"),
        availability_columns=[
            AvailabilityColumnMapping(
                header="Ven matin", date=dt.date(2026, 7, 10),
                day_part_code="NIGHT",
            ),
            AvailabilityColumnMapping(
                header="Ven aprem", date=dt.date(2030, 1, 1),
                day_part_code="MORNING",
            ),
        ],
    )
    errors = validate_mapping(_table([]), mapping, edition, day_parts)
    codes = {e.code for e in errors}
    assert "import.mapping.unknown_day_part" in codes
    assert "import.mapping.date_outside_edition" in codes