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 AvailabilityWindow, 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,
    *,
    availability=(),
    wished_position_ids=frozenset(),
    wished_companion_ids=frozenset(),
    wished_artist_windows=(),
    existing_shifts=(),
    existing_minutes=0,
) -> SolverVolunteer:
    return SolverVolunteer(
        id=vid,
        availability=tuple(availability),
        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 _full_day(day: int) -> AvailabilityWindow:
    return AvailabilityWindow(dt.date(2026, 7, day), dt.time(6, 0), dt.time(4, 0))


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", availability=[_full_day(10)])
    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",
        availability=[_full_day(10)],
        existing_shifts=[ShiftInfo(id="e", starts_at=_dt(10, 15), ends_at=_dt(10, 19))],
    )
    free = _volunteer("free", availability=[_full_day(10)])
    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", availability=[_full_day(10)])
    b = _volunteer("b", availability=[_full_day(10)])
    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", availability=[_full_day(10)])
    wisher = _volunteer(
        "wisher", availability=[_full_day(10)], wished_position_ids={"regie"}
    )
    solution = solve(SolverInput(shifts=(shift,), volunteers=(plain, wisher)))
    assert _by_shift(solution, "s1") == ("wisher",)


def test_prefers_fully_available_volunteer_over_partially_available():
    shift = _shift(start=_dt(10, 14), end=_dt(10, 18))
    full = _volunteer("full", availability=[_full_day(10)])
    partial = _volunteer(
        "partial",
        availability=[AvailabilityWindow(dt.date(2026, 7, 10), dt.time(14, 0), dt.time(16, 0))],
    )
    solution = solve(SolverInput(shifts=(shift,), volunteers=(full, partial)))
    assert _by_shift(solution, "s1") == ("full",)


def test_ranks_both_candidates_when_two_slots_are_open():
    shift = _shift(ideal_staff=2)
    full = _volunteer("full", availability=[_full_day(10)])
    partial = _volunteer(
        "partial",
        availability=[AvailabilityWindow(dt.date(2026, 7, 10), dt.time(14, 0), dt.time(16, 0))],
    )
    solution = solve(SolverInput(shifts=(shift,), volunteers=(full, partial)))
    assert _by_shift(solution, "s1") == ("full", "partial")


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", availability=[_full_day(10)])
    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_avoids_volunteer_with_insufficient_rest_since_last_shift():
    # < 8 h entre la fin d'un créneau déjà décidé et le début de celui-ci.
    shift = _shift(start=_dt(10, 14), end=_dt(10, 18))
    tired = _volunteer(
        "tired",
        availability=[_full_day(10)],
        existing_shifts=[ShiftInfo(id="e", starts_at=_dt(10, 8), ends_at=_dt(10, 10))],  # repos 4 h
    )
    fresh = _volunteer("fresh", availability=[_full_day(10)])
    solution = solve(SolverInput(shifts=(shift,), volunteers=(tired, fresh)))
    assert _by_shift(solution, "s1") == ("fresh",)


def test_daily_cumulative_hours_no_longer_penalized():
    # Le solveur ne plafonne plus le cumul de 8 h/jour (retiré) : seule la
    # validation, à la confirmation, le signale désormais
    # (`DAILY_HOURS_EXCEEDED`). Repos respecté (8 h pile) ; cumul du jour
    # (7 h + 2 h = 9 h) aurait auparavant été pénalisé.
    shift = _shift(ideal_staff=2, start=_dt(10, 21), end=_dt(10, 23))
    long_day = _volunteer(
        "long_day",
        availability=[_full_day(10)],
        existing_shifts=[ShiftInfo(id="e", starts_at=_dt(10, 6), ends_at=_dt(10, 13))],
    )
    fresh = _volunteer("fresh", availability=[_full_day(10)])
    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():
    # Repos suffisant (8 h pile) entre les deux créneaux : 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",
        availability=[_full_day(10)],
        existing_shifts=[ShiftInfo(id="prev", starts_at=_dt(10, 8), ends_at=_dt(10, 12))],
    )
    fresh = _volunteer("fresh", availability=[_full_day(10)])
    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",
        availability=[_full_day(10)],
        existing_shifts=[ShiftInfo(id="prev", starts_at=_dt(10, 8), ends_at=_dt(10, 12))],
    )
    fresh = _volunteer("fresh", availability=[_full_day(10)])
    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",
        availability=[_full_day(10)],
        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", availability=[_full_day(10)], existing_shifts=[unrelated]
    )
    truly_fresh = _volunteer("truly_fresh", availability=[_full_day(10)])
    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", availability=[_full_day(10)], wished_artist_windows=[passage]
    )
    indiff = _volunteer("indiff", availability=[_full_day(10)])
    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", availability=[_full_day(10)], 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", availability=[_full_day(10)], 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", availability=[_full_day(10)], wished_companion_ids={"copain"}
    )
    stranger = _volunteer("stranger", availability=[_full_day(10)])
    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", availability=[_full_day(10)], wished_companion_ids={"a", "b"}
    )
    one_friend = _volunteer(
        "one_friend", availability=[_full_day(10)], 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", availability=[_full_day(10)], wished_companion_ids={"not-here"}
    )
    solution = solve(SolverInput(shifts=(shift,), volunteers=(volunteer,)))
    assert _by_shift(solution, "s1") == ("v",)