benevoles/backend/tests/test_api_volunteers.py

import datetime as dt
import io
import uuid

from openpyxl import load_workbook
from pypdf import PdfReader
from sqlmodel import select

from app.models import Assignment, Availability, FeePaymentStatus, Role, Volunteer
from tests.factories import (
    dt_at,
    make_artist,
    make_availability,
    make_position,
    make_shift,
    make_volunteer,
)


def _url(edition, suffix=""):
    return f"/editions/{edition.id}/volunteers{suffix}"


def test_create_volunteer(client, edition):
    resp = client.post(
        _url(edition),
        json={
            "first_name": "Alex",
            "last_name": "Martin",
            "email": "alex@ex.fr",
        },
    )
    assert resp.status_code == 201
    body = resp.json()
    assert body["edition_id"] == str(edition.id)
    assert body["artist_ids"] == []
    assert body["wished_position_ids"] == []


def test_several_volunteers_can_share_an_email(client, edition):
    # Un couple, une famille : même boîte mail, deux bénévoles distincts.
    payload = {"first_name": "A", "last_name": "A", "email": "dup@ex.fr"}
    first = client.post(_url(edition), json=payload)
    assert first.status_code == 201
    second = client.post(_url(edition), json={**payload, "first_name": "B"})
    assert second.status_code == 201
    assert first.json()["id"] != second.json()["id"]


def test_volunteers_without_email_are_allowed(client, edition):
    for name in ("Un", "Deux"):
        r = client.post(
            _url(edition), json={"first_name": name, "last_name": "SansMail"}
        )
        assert r.status_code == 201


def test_list_is_ordered_by_name(client, edition, session):
    make_volunteer(session, edition, first_name="Bob", last_name="Zoe")
    make_volunteer(session, edition, first_name="Ana", last_name="Ada")
    names = [v["last_name"] for v in client.get(_url(edition)).json()]
    assert names == ["Ada", "Zoe"]


def test_get_unknown_id_is_404(client, edition):
    assert client.get(_url(edition, f"/{uuid.uuid4()}")).status_code == 404


def test_get_volunteer_of_another_edition_is_404(client, edition, session):
    from app.models import Edition

    other = Edition(
        year=2027,
        start_date=dt.date(2027, 7, 1),
        end_date=dt.date(2027, 7, 4),
    )
    session.add(other)
    session.commit()
    session.refresh(other)
    stranger = make_volunteer(session, other, email="x@ex.fr")

    assert client.get(_url(edition, f"/{stranger.id}")).status_code == 404


def test_create_with_artist_and_position_wishes(client, edition, session):
    artist = make_artist(session, edition)
    position = make_position(session, edition)
    resp = client.post(
        _url(edition),
        json={
            "first_name": "Fan",
            "last_name": "Zele",
            "artist_ids": [str(artist.id)],
            "wished_position_ids": [str(position.id)],
        },
    )
    assert resp.status_code == 201
    body = resp.json()
    assert body["artist_ids"] == [str(artist.id)]
    assert body["wished_position_ids"] == [str(position.id)]


def test_unknown_artist_wish_is_422(client, edition):
    resp = client.post(
        _url(edition),
        json={
            "first_name": "A",
            "last_name": "B",
            "artist_ids": [str(uuid.uuid4())],
        },
    )
    assert resp.status_code == 422
    assert resp.json()["detail"]["code"] == "artist.unknown"


def test_create_with_companion_wish(client, edition, session):
    friend = make_volunteer(session, edition, email="friend@ex.fr")
    resp = client.post(
        _url(edition),
        json={
            "first_name": "Alex",
            "last_name": "Martin",
            "wished_companion_ids": [str(friend.id)],
        },
    )
    assert resp.status_code == 201
    assert resp.json()["wished_companion_ids"] == [str(friend.id)]


def test_unknown_companion_wish_is_422(client, edition):
    resp = client.post(
        _url(edition),
        json={
            "first_name": "A",
            "last_name": "B",
            "wished_companion_ids": [str(uuid.uuid4())],
        },
    )
    assert resp.status_code == 422
    assert resp.json()["detail"]["code"] == "volunteer.unknown"


def test_cannot_wish_oneself_as_companion(client, edition, session):
    volunteer = make_volunteer(session, edition, email="solo@ex.fr")
    resp = client.patch(
        _url(edition, f"/{volunteer.id}"),
        json={"wished_companion_ids": [str(volunteer.id)]},
    )
    assert resp.status_code == 422
    assert resp.json()["detail"]["code"] == "volunteer.cannot_wish_self"


def test_companion_wish_is_one_directional(client, edition, session):
    # A souhaite B ne veut pas dire que B souhaite A.
    a = make_volunteer(session, edition, email="a@ex.fr")
    b = make_volunteer(session, edition, email="b@ex.fr")
    client.patch(_url(edition, f"/{a.id}"), json={"wished_companion_ids": [str(b.id)]})
    assert client.get(_url(edition, f"/{a.id}")).json()["wished_companion_ids"] == [str(b.id)]
    assert client.get(_url(edition, f"/{b.id}")).json()["wished_companion_ids"] == []


def test_deleting_a_wished_companion_clears_the_wish(client, edition, session):
    # RGPD : effacement complet, y compris les vœux dont on est la cible.
    a = make_volunteer(session, edition, email="a@ex.fr")
    b = make_volunteer(session, edition, email="b@ex.fr")
    client.patch(_url(edition, f"/{a.id}"), json={"wished_companion_ids": [str(b.id)]})
    assert client.delete(_url(edition, f"/{b.id}")).status_code == 204
    assert client.get(_url(edition, f"/{a.id}")).json()["wished_companion_ids"] == []


def test_patch_replaces_and_clears_wishes(client, edition, session):
    a1 = make_artist(session, edition, name="A1")
    a2 = make_artist(session, edition, name="A2")
    volunteer = make_volunteer(session, edition, email="w@ex.fr")

    r1 = client.patch(
        _url(edition, f"/{volunteer.id}"),
        json={"artist_ids": [str(a2.id), str(a1.id), str(a2.id)]},
    )
    # Doublons retirés, tri par (date, nom).
    assert r1.json()["artist_ids"] == [str(a1.id), str(a2.id)]

    r2 = client.patch(
        _url(edition, f"/{volunteer.id}"), json={"artist_ids": []}
    )
    assert r2.json()["artist_ids"] == []
    # Champ absent = inchangé.
    r3 = client.patch(
        _url(edition, f"/{volunteer.id}"), json={"first_name": "Zoe"}
    )
    assert r3.json()["artist_ids"] == []


def test_patch_updates_simple_fields(client, edition, session):
    volunteer = make_volunteer(session, edition, email="p@ex.fr")
    r1 = client.patch(
        _url(edition, f"/{volunteer.id}"),
        json={"first_name": "Neo", "comment": "dispo tôt"},
    )
    assert r1.status_code == 200
    assert r1.json()["first_name"] == "Neo"
    assert r1.json()["comment"] == "dispo tôt"


def test_delete_is_a_full_cascade(client, edition, session):
    volunteer = make_volunteer(session, edition, email="gone@ex.fr")
    position = make_position(session, edition)
    shift = make_shift(
        session, edition, position, start=dt_at(10, 9), end=dt_at(10, 12)
    )
    make_availability(
        session,
        edition,
        volunteer,
        date=dt.date(2026, 7, 10),
        start_time=dt.time(8, 0),
        end_time=dt.time(13, 0),
    )
    session.add(
        Assignment(
            edition_id=edition.id,
            volunteer_id=volunteer.id,
            shift_id=shift.id,
        )
    )
    session.commit()

    resp = client.delete(_url(edition, f"/{volunteer.id}"))
    assert resp.status_code == 204
    assert client.get(_url(edition, f"/{volunteer.id}")).status_code == 404
    assert session.exec(select(Volunteer)).all() == []
    assert session.exec(select(Assignment)).all() == []
    assert session.exec(select(Availability)).all() == []
    # Le créneau subsiste.
    from app.models import Shift

    assert session.get(Shift, shift.id) is not None


# --------------------------------------------------------------------------- #
# Planning enrichi (poste, lieu, référent) — écran bénévole                    #
# --------------------------------------------------------------------------- #
def test_planning_is_empty_without_assignment(client, edition, session):
    volunteer = make_volunteer(session, edition, email="v@ex.fr")
    resp = client.get(_url(edition, f"/{volunteer.id}/planning"))
    assert resp.status_code == 200
    assert resp.json() == []


def test_planning_includes_position_and_referent(client, edition, session):
    from app.models import Account

    volunteer = make_volunteer(session, edition, email="v@ex.fr")
    referent_volunteer = make_volunteer(
        session, edition, first_name="Rita", last_name="Referente", email="r@ex.fr"
    )
    referent_volunteer.phone = "0600000000"
    session.add(referent_volunteer)
    referent_account = Account(
        edition_id=edition.id,
        email="rita@ex.fr",
        role=Role.LEAD,
        volunteer_id=referent_volunteer.id,
    )
    session.add(referent_account)
    session.commit()
    session.refresh(referent_account)

    position = make_position(session, edition, name="Bar central")
    position.location = "Chapiteau"
    position.lead_account_id = referent_account.id
    session.add(position)
    session.commit()

    shift = make_shift(
        session, edition, position, start=dt_at(10, 9), end=dt_at(10, 12)
    )
    session.add(
        Assignment(edition_id=edition.id, volunteer_id=volunteer.id, shift_id=shift.id)
    )
    session.commit()

    resp = client.get(_url(edition, f"/{volunteer.id}/planning"))
    assert resp.status_code == 200
    body = resp.json()
    assert len(body) == 1
    entry = body[0]
    assert entry["shift_id"] == str(shift.id)
    assert entry["position_name"] == "Bar central"
    assert entry["position_location"] == "Chapiteau"
    assert entry["referent_name"] == "Rita Referente"
    assert entry["referent_phone"] == "0600000000"


def test_planning_referent_fields_are_none_without_referent(
    client, edition, session
):
    volunteer = make_volunteer(session, edition, email="v@ex.fr")
    position = make_position(session, edition)
    shift = make_shift(
        session, edition, position, start=dt_at(10, 9), end=dt_at(10, 12)
    )
    session.add(
        Assignment(edition_id=edition.id, volunteer_id=volunteer.id, shift_id=shift.id)
    )
    session.commit()

    resp = client.get(_url(edition, f"/{volunteer.id}/planning"))
    entry = resp.json()[0]
    assert entry["referent_name"] is None
    assert entry["referent_phone"] is None


def test_planning_referent_fields_are_none_when_lead_has_no_volunteer_record(
    client, edition, session
):
    """Un référent peut être un compte LEAD sans fiche bénévole associée :
    pas de nom/téléphone à afficher dans ce cas."""
    from app.models import Account

    volunteer = make_volunteer(session, edition, email="v@ex.fr")
    referent_account = Account(
        edition_id=edition.id, email="lead@ex.fr", role=Role.LEAD
    )
    session.add(referent_account)
    session.commit()
    session.refresh(referent_account)

    position = make_position(session, edition)
    position.lead_account_id = referent_account.id
    session.add(position)
    session.commit()

    shift = make_shift(
        session, edition, position, start=dt_at(10, 9), end=dt_at(10, 12)
    )
    session.add(
        Assignment(edition_id=edition.id, volunteer_id=volunteer.id, shift_id=shift.id)
    )
    session.commit()

    resp = client.get(_url(edition, f"/{volunteer.id}/planning"))
    entry = resp.json()[0]
    assert entry["referent_name"] is None
    assert entry["referent_phone"] is None


def test_planning_is_ordered_by_start_time(client, edition, session):
    volunteer = make_volunteer(session, edition, email="v@ex.fr")
    position = make_position(session, edition)
    late = make_shift(
        session, edition, position, start=dt_at(11, 14), end=dt_at(11, 18)
    )
    early = make_shift(
        session, edition, position, start=dt_at(10, 9), end=dt_at(10, 12)
    )
    session.add_all(
        [
            Assignment(edition_id=edition.id, volunteer_id=volunteer.id, shift_id=late.id),
            Assignment(edition_id=edition.id, volunteer_id=volunteer.id, shift_id=early.id),
        ]
    )
    session.commit()

    resp = client.get(_url(edition, f"/{volunteer.id}/planning"))
    body = resp.json()
    assert [e["shift_id"] for e in body] == [str(early.id), str(late.id)]


def test_planning_is_404_for_unknown_volunteer(client, edition):
    resp = client.get(_url(edition, f"/{uuid.uuid4()}/planning"))
    assert resp.status_code == 404


def test_volunteer_can_read_own_planning_but_not_anothers(
    session, edition, make_account, as_account
):
    me = make_volunteer(session, edition, email="me@e.fr")
    other = make_volunteer(session, edition, email="other@e.fr")
    account = make_account(Role.VOLUNTEER, volunteer_id=me.id, email="me@e.fr")
    client = as_account(account)

    assert client.get(_url(edition, f"/{me.id}/planning")).status_code == 200
    assert client.get(_url(edition, f"/{other.id}/planning")).status_code == 403


def test_lead_can_read_any_volunteers_planning(
    session, edition, make_account, as_account
):
    other = make_volunteer(session, edition, email="other@e.fr")
    lead_account = make_account(Role.LEAD)
    client = as_account(lead_account)

    assert client.get(_url(edition, f"/{other.id}/planning")).status_code == 200


def test_export_volunteers_as_excel(client, edition, session):
    position = make_position(session, edition, name="Bar central")
    artist = make_artist(session, edition, name="Chilla")
    volunteer = make_volunteer(
        session, edition, first_name="Alex", last_name="Martin", email="alex@ex.fr"
    )
    client.patch(
        _url(edition, f"/{volunteer.id}"),
        json={
            "wished_position_ids": [str(position.id)],
            "artist_ids": [str(artist.id)],
        },
    )

    resp = client.get(_url(edition, "/export"))
    assert resp.status_code == 200
    assert resp.headers["content-type"] == (
        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
    )
    assert f"benevoles-{edition.year}.xlsx" in resp.headers["content-disposition"]

    workbook = load_workbook(io.BytesIO(resp.content))
    sheet = workbook.active
    rows = list(sheet.iter_rows(values_only=True))
    assert rows[0][0] == "Prénom"
    row = rows[1]
    assert row[0] == "Alex"
    assert row[1] == "Martin"
    assert row[2] == "alex@ex.fr"
    assert row[5] == "Bar central"
    assert row[6] == "Chilla"


def test_export_neutralizes_formula_injection(client, edition, session):
    # CWE-1236 : un bénévole inscrit via le formulaire public (sans
    # authentification) contrôle son nom, son commentaire, etc. — ouverts
    # tels quels dans Excel par la coordination, une valeur comme
    # `=cmd|'/c calc'!A0` y serait exécutée comme une formule. Un préfixe
    # `'` force l'interprétation en texte (voir `app/exporting/xlsx.py`).
    volunteer = make_volunteer(
        session, edition, first_name="=cmd|'/c calc'!A0", last_name="Martin"
    )
    client.patch(
        _url(edition, f"/{volunteer.id}"),
        json={"comment": "@SUM(1+1)*cmd|' /C calc'!A0"},
    )

    resp = client.get(_url(edition, "/export"))
    workbook = load_workbook(io.BytesIO(resp.content))
    row = list(workbook.active.iter_rows(values_only=True))[1]
    assert row[0] == "'=cmd|'/c calc'!A0"
    assert row[17] == "'@SUM(1+1)*cmd|' /C calc'!A0"


def test_export_volunteers_is_coordinator_only(edition, make_account, as_account):
    lead = make_account(Role.LEAD, email="lead@e.fr")
    resp = as_account(lead).get(_url(edition, "/export"))
    assert resp.status_code == 403


def _mark_mecenat_paid(session, volunteer, *, amount=50.0):
    """Simule l'état d'un bénévole ayant payé un type de paiement « mécénat »
    (voir `app/api/forms.py:_promote_pending`) — `mecenat_address` est le
    marqueur qui distingue ce chemin d'un simple paiement classique."""
    volunteer.fee_payment_status = FeePaymentStatus.PAID
    volunteer.fee_amount = amount
    volunteer.fee_paid_at = dt.datetime(2026, 3, 5, 10, 0)
    volunteer.mecenat_address = "12 rue des Lilas"
    volunteer.mecenat_city = "Paris"
    volunteer.mecenat_postal_code = "75020"
    volunteer.mecenat_country = "France"
    session.add(volunteer)
    session.commit()


def test_mecenat_receipts_pdf_includes_only_paid_mecenat_donors(
    client, edition, session
):
    paid_mecenat = make_volunteer(
        session, edition, first_name="Alex", last_name="Martin"
    )
    _mark_mecenat_paid(session, paid_mecenat)

    # Cotisation classique payée (pas de mécénat) : ne doit pas apparaître.
    paid_classique = make_volunteer(
        session, edition, first_name="Sam", last_name="Petit"
    )
    paid_classique.fee_payment_status = FeePaymentStatus.PAID
    paid_classique.fee_amount = 15.0
    paid_classique.fee_paid_at = dt.datetime(2026, 3, 5, 10, 0)
    session.add(paid_classique)
    session.commit()

    # Mécénat choisi mais paiement non abouti : ne doit pas apparaître.
    failed_mecenat = make_volunteer(
        session, edition, first_name="Jo", last_name="Dupuis"
    )
    failed_mecenat.fee_payment_status = FeePaymentStatus.FAILED
    failed_mecenat.mecenat_address = "1 rue X"
    session.add(failed_mecenat)
    session.commit()

    resp = client.get(_url(edition, "/mecenat-receipts"))
    assert resp.status_code == 200
    assert resp.headers["content-type"] == "application/pdf"
    assert f"recus-mecenat-{edition.year}.pdf" in resp.headers["content-disposition"]

    reader = PdfReader(io.BytesIO(resp.content))
    # Un reçu = 2 pages (le cerfa officiel) — un seul bénévole qualifie ici.
    assert len(reader.pages) == 2
    donor_page_text = reader.pages[1].extract_text()
    assert "Martin" in donor_page_text
    assert "Alex" in donor_page_text
    assert "50,00" in donor_page_text
    assert "05/03/2026" in donor_page_text
    assert "Petit" not in donor_page_text


def test_mecenat_receipts_is_422_without_any_paid_donor(client, edition):
    resp = client.get(_url(edition, "/mecenat-receipts"))
    assert resp.status_code == 422
    assert resp.json()["detail"]["code"] == "mecenat.no_receipts"


def test_mecenat_receipts_is_coordinator_only(
    edition, session, make_account, as_account
):
    volunteer = make_volunteer(session, edition, first_name="Alex", last_name="Martin")
    _mark_mecenat_paid(session, volunteer)
    lead = make_account(Role.LEAD, email="lead@e.fr")
    resp = as_account(lead).get(_url(edition, "/mecenat-receipts"))
    assert resp.status_code == 403