benevoles/backend/tests/test_solver_engine.py

"""Tests du solveur (suggestions, jamais d'affectation automatique)."""

import datetime as dt

from app.solver.engine import solve
from app.solver.models import SolverInput, SolverShift, SolverVolunteer
from app.validation.models import ShiftInfo


def _dt(day: int, hour: int, minute: int = 0) -> dt.datetime:
    return dt.datetime(2026, 7, day, hour, minute)


def _shift(
    sid: str = "s1",
    *,
    position_id: str = "p1",
    start: dt.datetime | None = None,
    end: dt.datetime | None = None,
    min_staff=1,
    ideal_staff=1,
    already_assigned=0,
    predecessor_shift_ids=frozenset(),
    assigned_volunteer_ids=frozenset(),
) -> SolverShift:
    return SolverShift(
        id=sid,
        position_id=position_id,
        starts_at=start or _dt(10, 14),
        ends_at=end or _dt(10, 18),
        min_staff=min_staff,
        ideal_staff=ideal_staff,
        already_assigned=already_assigned,
        predecessor_shift_ids=frozenset(predecessor_shift_ids),
        assigned_volunteer_ids=frozenset(assigned_volunteer_ids),
    )


def _volunteer(
    vid: str,
    *,
    wished_position_ids=frozenset(),
    wished_companion_ids=frozenset(),
    wished_artist_windows=(),
    existing_shifts=(),
    existing_minutes=0,
) -> SolverVolunteer:
    return SolverVolunteer(
        id=vid,
        wished_position_ids=frozenset(wished_position_ids),
        wished_companion_ids=frozenset(wished_companion_ids),
        wished_artist_windows=tuple(wished_artist_windows),
        existing_shifts=tuple(existing_shifts),
        existing_minutes=existing_minutes,
    )


def _by_shift(solution, shift_id: str) -> tuple[str, ...]:
    for s in solution.suggestions:
        if s.shift_id == shift_id:
            return s.volunteer_ids
    return ()


def test_no_variables_returns_empty_solution():
    data = SolverInput(shifts=(), volunteers=())
    assert solve(data).suggestions == ()


def test_fully_staffed_shift_gets_no_suggestion():
    shift = _shift(ideal_staff=1, already_assigned=1)
    volunteer = _volunteer("v1")
    solution = solve(SolverInput(shifts=(shift,), volunteers=(volunteer,)))
    assert solution.suggestions == ()


def test_picks_the_only_free_volunteer_over_a_busy_one():
    shift = _shift()
    busy = _volunteer(
        "busy",
        existing_shifts=[ShiftInfo(id="e", starts_at=_dt(10, 15), ends_at=_dt(10, 19))],
    )
    free = _volunteer("free")
    solution = solve(SolverInput(shifts=(shift,), volunteers=(busy, free)))
    assert _by_shift(solution, "s1") == ("free",)


def test_does_not_exceed_ideal_staff_cap():
    shift = _shift(ideal_staff=2, already_assigned=1)  # il ne manque qu'une place
    a = _volunteer("a")
    b = _volunteer("b")
    solution = solve(SolverInput(shifts=(shift,), volunteers=(a, b)))
    assert len(_by_shift(solution, "s1")) == 1


def test_prefers_volunteer_with_matching_position_wish():
    shift = _shift(position_id="regie")
    plain = _volunteer("plain")
    wisher = _volunteer(
        "wisher", wished_position_ids={"regie"}
    )
    solution = solve(SolverInput(shifts=(shift,), volunteers=(plain, wisher)))
    assert _by_shift(solution, "s1") == ("wisher",)


def test_never_double_books_a_volunteer_across_two_new_suggestions():
    # Deux créneaux qui se chevauchent, un seul bénévole disponible pour les
    # deux : le solveur ne le propose qu'une fois.
    shift_a = _shift("a", start=_dt(10, 14), end=_dt(10, 18))
    shift_b = _shift("b", start=_dt(10, 16), end=_dt(10, 20))
    only = _volunteer("only")
    solution = solve(SolverInput(shifts=(shift_a, shift_b), volunteers=(only,)))
    picked_a = _by_shift(solution, "a")
    picked_b = _by_shift(solution, "b")
    assert len(picked_a) + len(picked_b) == 1


def test_short_rest_is_not_penalized():
    # Plus de repos minimal (8 h) : 4 h après un créneau déjà décidé, le
    # bénévole reste à égalité avec un autre.
    shift = _shift(ideal_staff=2, start=_dt(10, 14), end=_dt(10, 18))
    tired = _volunteer(
        "tired",
        existing_shifts=[ShiftInfo(id="e", starts_at=_dt(10, 8), ends_at=_dt(10, 10))],
    )
    fresh = _volunteer("fresh")
    solution = solve(SolverInput(shifts=(shift,), volunteers=(tired, fresh)))
    assert set(_by_shift(solution, "s1")) == {"tired", "fresh"}


def test_daily_cumulative_hours_no_longer_penalized():
    # Aucun plafond de cumul journalier : 7 h + 2 h = 9 h sur la journée ne
    # pénalise pas le bénévole.
    shift = _shift(ideal_staff=2, start=_dt(10, 21), end=_dt(10, 23))
    long_day = _volunteer(
        "long_day",
        existing_shifts=[ShiftInfo(id="e", starts_at=_dt(10, 6), ends_at=_dt(10, 13))],
    )
    fresh = _volunteer("fresh")
    solution = solve(SolverInput(shifts=(shift,), volunteers=(long_day, fresh)))
    assert set(_by_shift(solution, "s1")) == {"long_day", "fresh"}


def test_prefers_volunteer_without_the_immediately_preceding_shift():
    # Seul le chaînage doit différencier les deux bénévoles.
    shift = _shift(start=_dt(10, 20), end=_dt(10, 23), predecessor_shift_ids={"prev"})
    chained = _volunteer(
        "chained",
        existing_shifts=[ShiftInfo(id="prev", starts_at=_dt(10, 8), ends_at=_dt(10, 12))],
    )
    fresh = _volunteer("fresh")
    solution = solve(SolverInput(shifts=(shift,), volunteers=(chained, fresh)))
    assert _by_shift(solution, "s1") == ("fresh",)


def test_chain_penalty_applies_across_different_positions():
    # Le prédécesseur n'est pas limité au même poste : le bénévole enchaîne
    # ici depuis un autre poste, ça compte quand même — l'idée est de limiter
    # les enchaînements de shifts, pas de faire tourner un poste précis.
    shift = _shift(
        position_id="bar", start=_dt(10, 20), end=_dt(10, 23), predecessor_shift_ids={"prev"}
    )
    chained = _volunteer(
        "chained",
        existing_shifts=[ShiftInfo(id="prev", starts_at=_dt(10, 8), ends_at=_dt(10, 12))],
    )
    fresh = _volunteer("fresh")
    solution = solve(SolverInput(shifts=(shift,), volunteers=(chained, fresh)))
    assert _by_shift(solution, "s1") == ("fresh",)


def test_still_suggests_the_only_candidate_despite_chaining():
    # Le chaînage reste un avertissement, pas un blocage : personne d'autre
    # pour ce créneau urgent, le solveur le propose quand même.
    shift = _shift(start=_dt(10, 20), end=_dt(10, 23), predecessor_shift_ids={"prev"})
    chained = _volunteer(
        "chained",
        existing_shifts=[ShiftInfo(id="prev", starts_at=_dt(10, 8), ends_at=_dt(10, 12))],
    )
    solution = solve(SolverInput(shifts=(shift,), volunteers=(chained,)))
    assert _by_shift(solution, "s1") == ("chained",)


def test_an_unrelated_existing_shift_does_not_trigger_the_chain_penalty():
    # Le bénévole a bien un créneau existant, mais ce n'est pas l'un des
    # prédécesseurs désignés de ce créneau : pas de malus de chaînage (le
    # prédécesseur est identifié par id, pas par proximité). Les deux
    # candidats doivent donc être à égalité.
    shift = _shift(
        ideal_staff=2, start=_dt(10, 20), end=_dt(10, 23), predecessor_shift_ids={"prev"}
    )
    unrelated = ShiftInfo(id="not-the-predecessor", starts_at=_dt(10, 8), ends_at=_dt(10, 12))
    almost_chained = _volunteer(
        "almost_chained", existing_shifts=[unrelated]
    )
    truly_fresh = _volunteer("truly_fresh")
    solution = solve(SolverInput(shifts=(shift,), volunteers=(almost_chained, truly_fresh)))
    assert set(_by_shift(solution, "s1")) == {"almost_chained", "truly_fresh"}


def test_avoids_volunteer_whose_wished_artist_plays_during_the_shift():
    # Le créneau (14 h - 18 h) chevauche le passage de l'artiste (16 h - 17 h)
    # que « fan » veut voir : le solveur préfère « indiff », qui n'a pas ce
    # conflit.
    shift = _shift(start=_dt(10, 14), end=_dt(10, 18))
    passage = ShiftInfo(id="art1", starts_at=_dt(10, 16), ends_at=_dt(10, 17))
    fan = _volunteer(
        "fan", wished_artist_windows=[passage]
    )
    indiff = _volunteer("indiff")
    solution = solve(SolverInput(shifts=(shift,), volunteers=(fan, indiff)))
    assert _by_shift(solution, "s1") == ("indiff",)


def test_still_suggests_the_only_candidate_despite_an_artist_conflict():
    # Le conflit d'artiste reste un avertissement, pas un blocage : s'il n'y a
    # personne d'autre pour un créneau urgent, le solveur propose quand même.
    shift = _shift(start=_dt(10, 14), end=_dt(10, 18))
    passage = ShiftInfo(id="art1", starts_at=_dt(10, 16), ends_at=_dt(10, 17))
    fan = _volunteer(
        "fan", wished_artist_windows=[passage]
    )
    solution = solve(SolverInput(shifts=(shift,), volunteers=(fan,)))
    assert _by_shift(solution, "s1") == ("fan",)


def test_artist_conflict_outside_shift_hours_has_no_effect():
    # Le passage (20 h - 21 h) ne chevauche pas le créneau (14 h - 18 h) :
    # aucun malus, « fan » reste un candidat normal.
    shift = _shift(start=_dt(10, 14), end=_dt(10, 18))
    passage = ShiftInfo(id="art1", starts_at=_dt(10, 20), ends_at=_dt(10, 21))
    fan = _volunteer(
        "fan", wished_artist_windows=[passage]
    )
    solution = solve(SolverInput(shifts=(shift,), volunteers=(fan,)))
    assert _by_shift(solution, "s1") == ("fan",)


def test_prefers_volunteer_whose_wished_companion_is_already_on_the_shift():
    # « buddy » a souhaité être avec « copain », déjà affecté au créneau :
    # préféré à « stranger », qui n'a pas ce lien.
    shift = _shift(ideal_staff=2, already_assigned=1, assigned_volunteer_ids={"copain"})
    buddy = _volunteer(
        "buddy", wished_companion_ids={"copain"}
    )
    stranger = _volunteer("stranger")
    solution = solve(SolverInput(shifts=(shift,), volunteers=(buddy, stranger)))
    assert _by_shift(solution, "s1") == ("buddy",)


def test_companion_bonus_scales_with_number_of_matches():
    # Deux compagnons souhaités déjà présents valent mieux qu'un seul.
    shift = _shift(
        ideal_staff=4, already_assigned=2, assigned_volunteer_ids={"a", "b"}
    )
    two_friends = _volunteer(
        "two_friends", wished_companion_ids={"a", "b"}
    )
    one_friend = _volunteer(
        "one_friend", wished_companion_ids={"a"}
    )
    solution = solve(SolverInput(shifts=(shift,), volunteers=(two_friends, one_friend)))
    assert _by_shift(solution, "s1") == ("two_friends", "one_friend")


def test_wished_companion_absent_from_the_shift_has_no_effect():
    shift = _shift()
    volunteer = _volunteer(
        "v", wished_companion_ids={"not-here"}
    )
    solution = solve(SolverInput(shifts=(shift,), volunteers=(volunteer,)))
    assert _by_shift(solution, "s1") == ("v",)