benevoles/backend/tests/test_models.py

import datetime as dt

from sqlalchemy import text

from app.models import (
    Assignment,
    AssignmentSource,
    Availability,
    Edition,
    Position,
    PositionCategory,
    Shift,
    Volunteer,
)

EXPECTED_TABLES = {
    "edition",
    "day_part",
    "artist",
    "volunteer",
    "volunteer_artist",
    "volunteer_position_wish",
    "volunteer_companion_wish",
    "account",
    "magic_link_token",
    "position",
    "shift",
    "availability",
    "assignment",
    "badge_template",
    "badge_field",
    "form_template",
    "form_field",
    "form_payment_option",
    "pending_form_submission",
}


def _edition(session) -> Edition:
    e = Edition(
        year=2026,
        start_date=dt.date(2026, 7, 8),
        end_date=dt.date(2026, 7, 11),
        retention_days=365,
    )
    session.add(e)
    session.commit()
    session.refresh(e)
    return e


def test_metadata_has_expected_tables():
    from sqlmodel import SQLModel

    assert set(SQLModel.metadata.tables) == EXPECTED_TABLES


def test_create_core_graph(session):
    edition = _edition(session)

    volunteer = Volunteer(
        edition_id=edition.id,
        email="alex@example.org",
        first_name="Alex",
        last_name="Martin",
    )
    position = Position(
        edition_id=edition.id, name="Bar principal", category=PositionCategory.BAR
    )
    session.add(volunteer)
    session.add(position)
    session.commit()
    session.refresh(volunteer)
    session.refresh(position)

    shift = Shift(
        edition_id=edition.id,
        position_id=position.id,
        starts_at=dt.datetime(2026, 7, 9, 18, 0),
        ends_at=dt.datetime(2026, 7, 9, 23, 0),
        min_staff=2,
        ideal_staff=4,
    )
    session.add(shift)
    session.commit()
    session.refresh(volunteer)
    session.refresh(shift)

    assignment = Assignment(
        edition_id=edition.id, volunteer_id=volunteer.id, shift_id=shift.id
    )
    session.add(assignment)
    session.commit()
    session.refresh(assignment)

    # Valeurs par défaut appliquées.
    assert assignment.source == AssignmentSource.MANUAL
    assert assignment.locked is False
    # Relations.
    assert shift.position.id == position.id
    assert volunteer.assignments[0].id == assignment.id


def test_availability_window_round_trip(session):
    edition = _edition(session)
    volunteer = Volunteer(
        edition_id=edition.id, first_name="Sam", last_name="Nuit"
    )
    session.add(volunteer)
    session.commit()
    session.refresh(volunteer)

    avail = Availability(
        edition_id=edition.id,
        volunteer_id=volunteer.id,
        date=dt.date(2026, 7, 9),
        start_time=dt.time(19, 0),
        end_time=dt.time(2, 0),  # passe minuit
    )
    session.add(avail)
    session.commit()
    session.expire_all()

    reloaded = session.get(Availability, avail.id)
    assert reloaded.start_time == dt.time(19, 0)
    assert reloaded.end_time == dt.time(2, 0)


def test_volunteer_email_can_be_shared(session):
    # Un couple, une famille : même boîte mail, deux bénévoles distincts —
    # ce n'est plus une contrainte d'unicité en base (voir migration 0009).
    edition = _edition(session)
    session.add(
        Volunteer(
            edition_id=edition.id,
            email="dup@example.org",
            first_name="A",
            last_name="A",
        )
    )
    session.commit()
    session.add(
        Volunteer(
            edition_id=edition.id,
            email="dup@example.org",
            first_name="B",
            last_name="B",
        )
    )
    session.commit()  # ne lève pas


def test_multiple_volunteers_without_email_allowed(session):
    edition = _edition(session)
    session.add_all(
        [
            Volunteer(edition_id=edition.id, first_name="A", last_name="A"),
            Volunteer(edition_id=edition.id, first_name="B", last_name="B"),
        ]
    )
    session.commit()  # NULL n'entre pas dans la contrainte d'unicité


def test_deleting_volunteer_cascades_to_assignments(session):
    edition = _edition(session)
    volunteer = Volunteer(
        edition_id=edition.id, first_name="Cascade", last_name="Test"
    )
    position = Position(
        edition_id=edition.id, name="Accueil", category=PositionCategory.WELCOME
    )
    session.add_all([volunteer, position])
    session.commit()
    session.refresh(volunteer)
    session.refresh(position)

    shift = Shift(
        edition_id=edition.id,
        position_id=position.id,
        starts_at=dt.datetime(2026, 7, 9, 9, 0),
        ends_at=dt.datetime(2026, 7, 9, 12, 0),
    )
    session.add(shift)
    session.commit()
    session.refresh(shift)

    assignment = Assignment(
        edition_id=edition.id, volunteer_id=volunteer.id, shift_id=shift.id
    )
    session.add(assignment)
    session.commit()
    assignment_id = assignment.id
    volunteer_id = volunteer.id

    # Suppression au niveau base : la cascade ON DELETE doit s'appliquer.
    session.execute(
        text("DELETE FROM volunteer WHERE id = :id"), {"id": volunteer_id}
    )
    session.commit()
    session.expire_all()

    assert session.get(Assignment, assignment_id) is None
    # Le créneau, lui, subsiste.
    assert session.get(Shift, shift.id) is not None