Retrait des compétences ; disponibilités en fenêtres horaires libres

Compétences supprimées de bout en bout :
- tables `skill`, `volunteer_skill`, `shift_required_skill` ;
- `Volunteer.skill_codes`, `Shift.required_skill_codes`, colonne `skills`
  de l'import, règle de validation `MISSING_REQUIRED_SKILL`, jeu de démo.

Disponibilités repensées : `Availability` passe de (`day_part_id`,
`status` tri-état) à une **fenêtre horaire libre** (`date`, `start_time`,
`end_time` ; `end <= start` ⇒ passe minuit), plusieurs par jour.
- Nouveaux endpoints `GET`/`PUT /editions/{id}/volunteers/{vid}/availability`
  (PUT = remplacement complet, COORDINATION) et
  `GET /editions/{id}/availability` (toute l'édition, pour la grille).
- `OUTSIDE_AVAILABILITY` (toujours WARNING) : compare le créneau à l'union
  des fenêtres du bénévole ce jour-là ; renvoie `date` + `uncovered_minutes`.
  `DayPart` ne sert plus qu'à la borne de « journée » du cumul 8 h.
- L'import ne gère plus les disponibilités (fenêtres saisies dans l'app) :
  `values.py` supprimé, mapping réduit à nom/email/téléphone/commentaire.

Migration 0007 (recrée `availability`). `AvailabilityStatus` supprimé.
158 tests, `alembic check` et `ruff` au vert. Front à suivre.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017uUsRg7XzopkCPWimwg9ZM
This commit is contained in:
AdamThomas
2026-09-10 22:46:45 +02:00
co-authored by Claude Sonnet 5
parent 81609eecc5
commit be51770b51
36 changed files with 800 additions and 1199 deletions
+19 -53
View File
@@ -5,13 +5,10 @@ import datetime as dt
from app.models import (
Artist,
Availability,
AvailabilityStatus,
Position,
PositionCategory,
Shift,
Skill,
Volunteer,
VolunteerSkill,
)
@@ -36,7 +33,7 @@ def make_artist(
def make_shift(
session, edition, position, *, start, end, min_staff=1, ideal_staff=2, skills=()
session, edition, position, *, start, end, min_staff=1, ideal_staff=2
):
shift = Shift(
edition_id=edition.id,
@@ -47,16 +44,13 @@ def make_shift(
ideal_staff=ideal_staff,
)
session.add(shift)
session.flush()
for code in skills:
_link_skill(session, edition, code=code, shift=shift)
session.commit()
session.refresh(shift)
return shift
def make_volunteer(
session, edition, *, first_name="Alex", last_name="Martin", email=None, skills=()
session, edition, *, first_name="Alex", last_name="Martin", email=None
):
volunteer = Volunteer(
edition_id=edition.id,
@@ -65,66 +59,38 @@ def make_volunteer(
email=email,
)
session.add(volunteer)
session.flush()
for code in skills:
_link_skill(session, edition, volunteer=volunteer, code=code)
session.commit()
session.refresh(volunteer)
return volunteer
def make_availability(session, edition, volunteer, day_part, *, date, status):
def make_availability(
session, edition, volunteer, *, date, start_time, end_time
):
"""Fenêtre de disponibilité libre. `end_time <= start_time` ⇒ passe minuit."""
row = Availability(
edition_id=edition.id,
volunteer_id=volunteer.id,
day_part_id=day_part.id,
date=date,
status=status,
start_time=start_time,
end_time=end_time,
)
session.add(row)
session.commit()
return row
def declare_available_all_day(session, edition, volunteer, day_parts, date):
for part in day_parts.values():
make_availability(
session,
edition,
volunteer,
part,
date=date,
status=AvailabilityStatus.AVAILABLE,
)
def _get_or_create_skill(session, edition, code) -> Skill:
from sqlmodel import select
skill = session.exec(
select(Skill)
.where(Skill.edition_id == edition.id)
.where(Skill.code == code)
).first()
if skill is None:
skill = Skill(edition_id=edition.id, code=code, label=code)
session.add(skill)
session.flush()
return skill
def _link_skill(session, edition, *, code, volunteer=None, shift=None):
from app.models import ShiftRequiredSkill
skill = _get_or_create_skill(session, edition, code)
if volunteer is not None:
session.add(
VolunteerSkill(volunteer_id=volunteer.id, skill_id=skill.id)
)
if shift is not None:
session.add(
ShiftRequiredSkill(shift_id=shift.id, skill_id=skill.id)
)
def declare_available_all_day(session, edition, volunteer, date):
"""Disponible de 00:00 à 23:59 le jour donné (couvre tout un créneau
diurne ; pour la nuit, déclarer une fenêtre dédiée)."""
make_availability(
session,
edition,
volunteer,
date=date,
start_time=dt.time(0, 0),
end_time=dt.time(23, 59),
)
def dt_at(day: int, hour: int, minute: int = 0) -> dt.datetime:
+13 -26
View File
@@ -13,21 +13,21 @@ def _url(edition, suffix=""):
return f"/editions/{edition.id}/assignments{suffix}"
def _setup(session, edition, day_parts, *, available=True, skills=()):
def _setup(session, edition, *, available=True):
position = make_position(session, edition)
shift = make_shift(
session, edition, position, start=dt_at(10, 14), end=dt_at(10, 18)
)
volunteer = make_volunteer(session, edition, email="v@ex.fr", skills=skills)
volunteer = make_volunteer(session, edition, email="v@ex.fr")
if available:
declare_available_all_day(
session, edition, volunteer, day_parts, dt.date(2026, 7, 10)
session, edition, volunteer, dt.date(2026, 7, 10)
)
return volunteer, shift
def test_create_assignment_clean(client, edition, session, day_parts):
volunteer, shift = _setup(session, edition, day_parts)
volunteer, shift = _setup(session, edition)
resp = client.post(
_url(edition),
json={"volunteer_id": str(volunteer.id), "shift_id": str(shift.id)},
@@ -40,32 +40,19 @@ def test_create_assignment_clean(client, edition, session, day_parts):
def test_create_assignment_returns_warnings(client, edition, session, day_parts):
# Bénévole non déclaré disponible + compétence requise absente.
volunteer, shift = _setup(session, edition, day_parts, available=False)
position = make_position(session, edition, name="Régie")
shift_with_skill = make_shift(
session,
edition,
position,
start=dt_at(10, 14),
end=dt_at(10, 18),
skills=["SOUND"],
)
# Bénévole non déclaré disponible sur le créneau.
volunteer, shift = _setup(session, edition, available=False)
resp = client.post(
_url(edition),
json={
"volunteer_id": str(volunteer.id),
"shift_id": str(shift_with_skill.id),
},
json={"volunteer_id": str(volunteer.id), "shift_id": str(shift.id)},
)
assert resp.status_code == 201
codes = {w["code"] for w in resp.json()["warnings"]}
assert "MISSING_REQUIRED_SKILL" in codes
assert "OUTSIDE_AVAILABILITY" in codes
def test_create_assignment_blocked_by_overlap(client, edition, session, day_parts):
volunteer, shift = _setup(session, edition, day_parts)
volunteer, shift = _setup(session, edition)
assert client.post(
_url(edition),
json={"volunteer_id": str(volunteer.id), "shift_id": str(shift.id)},
@@ -94,7 +81,7 @@ def test_create_assignment_blocked_by_overlap(client, edition, session, day_part
def test_create_duplicate_assignment_conflicts(client, edition, session, day_parts):
volunteer, shift = _setup(session, edition, day_parts)
volunteer, shift = _setup(session, edition)
body = {"volunteer_id": str(volunteer.id), "shift_id": str(shift.id)}
assert client.post(_url(edition), json=body).status_code == 201
resp = client.post(_url(edition), json=body)
@@ -112,7 +99,7 @@ def test_assignment_wrong_edition_is_422(client, edition, session, day_parts):
session.commit()
session.refresh(other)
stranger = make_volunteer(session, other, email="s@ex.fr")
_, shift = _setup(session, edition, day_parts)
_, shift = _setup(session, edition)
resp = client.post(
_url(edition),
@@ -123,7 +110,7 @@ def test_assignment_wrong_edition_is_422(client, edition, session, day_parts):
def test_validate_endpoint_does_not_persist(client, edition, session, day_parts):
volunteer, shift = _setup(session, edition, day_parts, available=False)
volunteer, shift = _setup(session, edition, available=False)
resp = client.post(
_url(edition, "/validate"),
json={"volunteer_id": str(volunteer.id), "shift_id": str(shift.id)},
@@ -138,7 +125,7 @@ def test_validate_endpoint_does_not_persist(client, edition, session, day_parts)
def test_patch_locked_keeps_assignment_and_returns_warnings(
client, edition, session, day_parts
):
volunteer, shift = _setup(session, edition, day_parts, available=False)
volunteer, shift = _setup(session, edition, available=False)
aid = client.post(
_url(edition),
json={"volunteer_id": str(volunteer.id), "shift_id": str(shift.id)},
@@ -153,7 +140,7 @@ def test_patch_locked_keeps_assignment_and_returns_warnings(
def test_delete_assignment(client, edition, session, day_parts):
volunteer, shift = _setup(session, edition, day_parts)
volunteer, shift = _setup(session, edition)
aid = client.post(
_url(edition),
json={"volunteer_id": str(volunteer.id), "shift_id": str(shift.id)},
+114
View File
@@ -0,0 +1,114 @@
from app.models import Role
from tests.factories import make_volunteer
def _url(edition, volunteer, suffix=""):
return f"/editions/{edition.id}/volunteers/{volunteer.id}/availability{suffix}"
def test_put_then_get_returns_windows_sorted(client, edition, session):
v = make_volunteer(session, edition, email="v@e.fr")
resp = client.put(
_url(edition, v),
json={
"windows": [
{"date": "2026-07-11", "start_time": "14:00", "end_time": "19:00"},
{"date": "2026-07-10", "start_time": "08:00", "end_time": "13:00"},
{"date": "2026-07-10", "start_time": "19:00", "end_time": "02:00"},
]
},
)
assert resp.status_code == 200
got = [(w["date"], w["start_time"], w["end_time"]) for w in resp.json()]
assert got == [
("2026-07-10", "08:00:00", "13:00:00"),
("2026-07-10", "19:00:00", "02:00:00"),
("2026-07-11", "14:00:00", "19:00:00"),
]
assert client.get(_url(edition, v)).json() == resp.json()
def test_put_replaces_the_whole_set_and_dedupes(client, edition, session):
v = make_volunteer(session, edition, email="v@e.fr")
client.put(
_url(edition, v),
json={
"windows": [
{"date": "2026-07-10", "start_time": "08:00", "end_time": "12:00"}
]
},
)
resp = client.put(
_url(edition, v),
json={
"windows": [
{"date": "2026-07-11", "start_time": "09:00", "end_time": "17:00"},
{"date": "2026-07-11", "start_time": "09:00", "end_time": "17:00"},
]
},
)
body = resp.json()
assert len(body) == 1
assert body[0]["date"] == "2026-07-11"
def test_put_rejects_date_outside_edition(client, edition, session):
v = make_volunteer(session, edition, email="v@e.fr")
resp = client.put(
_url(edition, v),
json={
"windows": [
{"date": "2026-08-01", "start_time": "08:00", "end_time": "12:00"}
]
},
)
assert resp.status_code == 422
assert resp.json()["detail"]["code"] == "availability.date_outside_edition"
def test_put_rejects_empty_window(client, edition, session):
v = make_volunteer(session, edition, email="v@e.fr")
resp = client.put(
_url(edition, v),
json={
"windows": [
{"date": "2026-07-10", "start_time": "09:00", "end_time": "09:00"}
]
},
)
assert resp.status_code == 422
assert resp.json()["detail"]["code"] == "availability.empty_window"
def test_edition_wide_listing(client, edition, session):
v1 = make_volunteer(session, edition, first_name="A", email="a@e.fr")
v2 = make_volunteer(session, edition, first_name="B", email="b@e.fr")
for v in (v1, v2):
client.put(
_url(edition, v),
json={
"windows": [
{"date": "2026-07-10", "start_time": "08:00", "end_time": "13:00"}
]
},
)
body = client.get(f"/editions/{edition.id}/availability").json()
assert {w["volunteer_id"] for w in body} == {str(v1.id), str(v2.id)}
def test_lead_can_read_but_not_write(edition, session, make_account, as_account):
v = make_volunteer(session, edition, email="v@e.fr")
lead = make_account(Role.LEAD, email="lead@e.fr")
api = as_account(lead)
assert api.get(_url(edition, v)).status_code == 200
resp = api.put(_url(edition, v), json={"windows": []})
assert resp.status_code == 403
def test_volunteer_reads_only_their_own(edition, session, 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="acc@e.fr")
api = as_account(account)
assert api.get(_url(edition, me)).status_code == 200
assert api.get(_url(edition, other)).status_code == 403
+3 -18
View File
@@ -1,9 +1,5 @@
import datetime as dt
from sqlmodel import select
from app.models import Availability, AvailabilityStatus, Role, Shift
from tests.factories import dt_at, make_position, make_shift, make_volunteer
from app.models import Role, Shift
from tests.factories import dt_at, make_position, make_shift
def _url(edition, suffix=""):
@@ -46,29 +42,18 @@ def test_update_day_part(client, edition):
assert resp.json()["end_time"] == "12:00:00"
def test_delete_unlinks_shifts_and_removes_availabilities(
client, edition, session, day_parts
):
def test_delete_sets_shift_day_part_to_null(client, edition, session, day_parts):
morning = day_parts["MORNING"]
position = make_position(session, edition)
shift = make_shift(
session, edition, position, start=dt_at(10, 6), end=dt_at(10, 13)
)
shift.day_part_id = morning.id
volunteer = make_volunteer(session, edition, email="v@e.fr")
session.add(
Availability(
edition_id=edition.id, volunteer_id=volunteer.id,
date=dt.date(2026, 7, 10), day_part_id=morning.id,
status=AvailabilityStatus.AVAILABLE,
)
)
session.commit()
assert client.delete(_url(edition, f"/{morning.id}")).status_code == 204
session.expire_all()
assert session.get(Shift, shift.id).day_part_id is None
assert session.exec(select(Availability)).all() == []
def test_write_is_coordinator_only(edition, make_account, as_account):
+3 -9
View File
@@ -14,18 +14,16 @@ def _payload(position, **overrides):
"ends_at": "2026-07-10T22:00:00",
"min_staff": 2,
"ideal_staff": 4,
"required_skill_codes": ["SST"],
}
base.update(overrides)
return base
def test_create_shift_with_required_skills(client, edition, session):
def test_create_shift(client, edition, session):
position = make_position(session, edition)
resp = client.post(_url(edition), json=_payload(position))
assert resp.status_code == 201
body = resp.json()
assert body["required_skill_codes"] == ["SST"]
assert body["min_staff"] == 2
assert body["position_id"] == str(position.id)
@@ -75,20 +73,16 @@ def test_list_filter_by_position(client, edition, session):
assert len(client.get(_url(edition)).json()) == 2
def test_patch_shift_times_and_skills(client, edition, session):
def test_patch_shift_times(client, edition, session):
position = make_position(session, edition)
sid = client.post(_url(edition), json=_payload(position)).json()["id"]
patched = client.patch(
_url(edition, f"/{sid}"),
json={
"ends_at": "2026-07-10T23:30:00",
"required_skill_codes": ["SOUND", "SST"],
},
json={"ends_at": "2026-07-10T23:30:00"},
)
assert patched.status_code == 200
assert patched.json()["ends_at"] == "2026-07-10T23:30:00"
assert patched.json()["required_skill_codes"] == ["SOUND", "SST"]
def test_patch_shift_into_invalid_range_is_422(client, edition, session):
+10 -15
View File
@@ -3,7 +3,7 @@ import uuid
from sqlmodel import select
from app.models import Assignment, Availability, AvailabilityStatus, Volunteer
from app.models import Assignment, Availability, Volunteer
from tests.factories import (
dt_at,
make_artist,
@@ -18,20 +18,20 @@ def _url(edition, suffix=""):
return f"/editions/{edition.id}/volunteers{suffix}"
def test_create_volunteer_with_skills(client, edition):
def test_create_volunteer(client, edition):
resp = client.post(
_url(edition),
json={
"first_name": "Alex",
"last_name": "Martin",
"email": "alex@ex.fr",
"skill_codes": ["SST", "PERMIS_B", "SST"],
},
)
assert resp.status_code == 201
body = resp.json()
assert body["skill_codes"] == ["PERMIS_B", "SST"]
assert body["edition_id"] == str(edition.id)
assert body["artist_ids"] == []
assert body["wished_position_ids"] == []
def test_create_duplicate_email_conflicts(client, edition):
@@ -131,23 +131,18 @@ def test_patch_replaces_and_clears_wishes(client, edition, session):
assert r3.json()["artist_ids"] == []
def test_patch_replaces_skills_and_can_clear(client, edition, session):
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", "skill_codes": ["SST"]},
json={"first_name": "Neo", "comment": "dispo tôt"},
)
assert r1.status_code == 200
assert r1.json()["first_name"] == "Neo"
assert r1.json()["skill_codes"] == ["SST"]
r2 = client.patch(
_url(edition, f"/{volunteer.id}"), json={"skill_codes": []}
)
assert r2.json()["skill_codes"] == []
assert r1.json()["comment"] == "dispo tôt"
def test_delete_is_a_full_cascade(client, edition, session, day_parts):
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(
@@ -157,9 +152,9 @@ def test_delete_is_a_full_cascade(client, edition, session, day_parts):
session,
edition,
volunteer,
day_parts["MORNING"],
date=dt.date(2026, 7, 10),
status=AvailabilityStatus.AVAILABLE,
start_time=dt.time(8, 0),
end_time=dt.time(13, 0),
)
session.add(
Assignment(
+10 -31
View File
@@ -6,9 +6,9 @@ from sqlmodel import select
from app.models import Volunteer
_CSV = (
"Prénom;Nom;Email;Compétences;Ven matin\n"
"Alex;Martin;alex@ex.fr;SST, Permis B;x\n"
"Sam;Nguyen;sam@ex.fr;SST;si besoin\n"
"Prénom;Nom;Email;Téléphone\n"
"Alex;Martin;alex@ex.fr;0600000000\n"
"Sam;Nguyen;sam@ex.fr;\n"
).encode()
@@ -20,20 +20,13 @@ def _mapping(dry_run: bool) -> str:
"first_name": "Prénom",
"last_name": "Nom",
"email": "Email",
"skills": "Compétences",
"phone": "Téléphone",
},
"availability_columns": [
{
"header": "Ven matin",
"date": "2026-07-10",
"day_part_code": "MORNING",
}
],
}
)
def test_analyze_returns_headers_preview_and_day_parts(client, edition):
def test_analyze_returns_headers_preview_and_suggestion(client, edition):
resp = client.post(
f"/editions/{edition.id}/import/analyze",
files={"file": ("benevoles.csv", _CSV, "text/csv")},
@@ -41,16 +34,11 @@ def test_analyze_returns_headers_preview_and_day_parts(client, edition):
assert resp.status_code == 200
body = resp.json()
assert body["format"] == "csv"
assert body["headers"] == ["Prénom", "Nom", "Email", "Compétences", "Ven matin"]
assert body["headers"] == ["Prénom", "Nom", "Email", "Téléphone"]
assert body["row_count"] == 2
assert body["preview"][0]["values"]["Nom"] == "Martin"
assert body["suggested_mapping"]["first_name"] == "Prénom"
assert {d["code"] for d in body["day_parts"]} == {
"MORNING",
"AFTERNOON",
"EVENING",
}
assert body["edition_dates"][0] == "2026-07-09"
assert body["suggested_mapping"]["phone"] == "Téléphone"
def test_execute_dry_run_then_real(client, session, edition):
@@ -69,9 +57,7 @@ def test_execute_dry_run_then_real(client, session, edition):
data={"mapping": _mapping(dry_run=False)},
)
assert real.status_code == 200
report = real.json()
assert report["created"] == 2
assert sorted(report["skills_created"]) == ["PERMIS_B", "SST"]
assert real.json()["created"] == 2
assert len(session.exec(select(Volunteer)).all()) == 2
@@ -79,14 +65,7 @@ def test_execute_rejects_invalid_mapping(client, edition):
bad_mapping = json.dumps(
{
"dry_run": True,
"columns": {"first_name": "Prénom", "last_name": "Nom"},
"availability_columns": [
{
"header": "Ven matin",
"date": "2026-07-10",
"day_part_code": "NIGHT",
}
],
"columns": {"first_name": "Prénom", "last_name": "Absente"},
}
)
resp = client.post(
@@ -98,7 +77,7 @@ def test_execute_rejects_invalid_mapping(client, edition):
detail = resp.json()["detail"]
assert detail["code"] == "import.mapping.invalid"
assert any(
e["code"] == "import.mapping.unknown_day_part" for e in detail["errors"]
e["code"] == "import.mapping.column_not_found" for e in detail["errors"]
)
+22 -74
View File
@@ -1,17 +1,11 @@
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.schemas import ColumnMapping, ImportMapping
from app.importing.service import run_import, suggest_mapping, validate_mapping
from app.models import Availability, Skill, Volunteer
from app.models import Volunteer
HEADERS = ["Prénom", "Nom", "Email", "Taille", "Compétences", "Ven matin", "Ven aprem"]
HEADERS = ["Prénom", "Nom", "Email", "Téléphone", "Remarque"]
def _mapping(dry_run: bool = False) -> ImportMapping:
@@ -21,18 +15,9 @@ def _mapping(dry_run: bool = False) -> ImportMapping:
first_name="Prénom",
last_name="Nom",
email="Email",
skills="Compétences",
phone="Téléphone",
comment="Remarque",
),
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",
),
],
)
@@ -58,46 +43,36 @@ def test_suggest_mapping_matches_common_headers():
assert suggestion["phone"] == "Numéro de portable"
def test_happy_path_creates_volunteers_skills_and_availabilities(session, edition):
def test_happy_path_creates_volunteers(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", ""],
["Alex", "Martin", "alex@ex.fr", "0600000000", "arrive tôt"],
["Sam", "Nguyen", "sam@ex.fr", "", ""],
]
),
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 {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"}
assert volunteers["alex@ex.fr"].phone == "0600000000"
assert volunteers["alex@ex.fr"].comment == "arrive tôt"
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"]]),
table=_table([["Alex", "Martin", "alex@ex.fr", "", ""]]),
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):
@@ -115,7 +90,7 @@ def test_existing_email_is_updated_not_duplicated(session, edition):
report = run_import(
session=session,
edition=edition,
table=_table([["Alex", "Martin", "ALEX@ex.fr", "M", "SST", "x", ""]]),
table=_table([["Alex", "Martin", "ALEX@ex.fr", "", ""]]),
mapping=_mapping(),
)
@@ -133,8 +108,8 @@ def test_duplicate_email_within_file_is_skipped(session, edition):
edition=edition,
table=_table(
[
["Alex", "Martin", "dup@ex.fr", "M", "", "x", ""],
["Alex", "Doublon", "dup@ex.fr", "L", "", "x", ""],
["Alex", "Martin", "dup@ex.fr", "", ""],
["Alex", "Doublon", "dup@ex.fr", "", ""],
]
),
mapping=_mapping(),
@@ -161,7 +136,7 @@ def test_row_without_email_matching_existing_name_warns_but_creates(
report = run_import(
session=session,
edition=edition,
table=_table([["jean", "DUPONT", "", "", "", "x", ""]]),
table=_table([["jean", "DUPONT", "", "", ""]]),
mapping=_mapping(),
)
assert report.created == 1
@@ -173,7 +148,7 @@ def test_missing_required_field_is_skipped(session, edition):
report = run_import(
session=session,
edition=edition,
table=_table([["", "Martin", "x@ex.fr", "", "", "x", ""]]),
table=_table([["", "Martin", "x@ex.fr", "", ""]]),
mapping=_mapping(),
)
assert report.skipped == 1
@@ -185,45 +160,18 @@ def test_invalid_email_is_skipped(session, edition):
report = run_import(
session=session,
edition=edition,
table=_table([["Alex", "Martin", "pas-un-email", "", "", "x", ""]]),
table=_table([["Alex", "Martin", "pas-un-email", "", ""]]),
mapping=_mapping(),
)
assert report.skipped == 1
assert report.rows[0].issues[0].code == "import.row.invalid_email"
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))
)
def test_validate_mapping_reports_missing_and_unknown_columns():
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",
),
],
columns=ColumnMapping(first_name="Prénom", last_name="", email="Absente")
)
errors = validate_mapping(_table([]), mapping, edition, day_parts)
errors = validate_mapping(_table([]), mapping)
codes = {e.code for e in errors}
assert "import.mapping.unknown_day_part" in codes
assert "import.mapping.date_outside_edition" in codes
assert "import.mapping.required_field_missing" in codes
assert "import.mapping.column_not_found" in codes
+12
View File
@@ -0,0 +1,12 @@
from app.importing.text import normalize, slug_code
def test_normalize_strips_accents_case_and_spaces():
assert normalize(" Régie SON ") == "regie son"
assert normalize("") == ""
def test_slug_code_is_stable_across_accents_and_case():
assert slug_code("Régie son") == "REGIE_SON"
assert slug_code(" apres midi ") == "APRES_MIDI"
assert slug_code("SST") == "SST"
-48
View File
@@ -1,48 +0,0 @@
import pytest
from app.importing.text import slug_code, split_skills
from app.importing.values import parse_availability_cell
from app.models import AvailabilityStatus
@pytest.mark.parametrize(
("cell", "expected"),
[
("x", AvailabilityStatus.AVAILABLE),
("OUI", AvailabilityStatus.AVAILABLE),
(" Disponible ", AvailabilityStatus.AVAILABLE),
("si besoin", AvailabilityStatus.IF_NEEDED),
("Peut-être", AvailabilityStatus.IF_NEEDED),
("non", AvailabilityStatus.UNAVAILABLE),
("0", AvailabilityStatus.UNAVAILABLE),
],
)
def test_recognized_availability_values(cell, expected):
status, recognized = parse_availability_cell(cell)
assert recognized is True
assert status is expected
def test_empty_cell_is_recognized_but_no_declaration():
assert parse_availability_cell(" ") == (None, True)
assert parse_availability_cell("N/A") == (None, True)
def test_unknown_cell_is_flagged():
assert parse_availability_cell("bof") == (None, False)
def test_slug_code_is_stable_across_accents_and_case():
assert slug_code("Régie son") == "REGIE_SON"
assert slug_code(" permis B ") == "PERMIS_B"
assert slug_code("SST") == "SST"
def test_split_skills():
assert split_skills("SST, Permis B / régie son\nlangues") == [
"SST",
"Permis B",
"régie son",
"langues",
]
assert split_skills("") == []
+6 -25
View File
@@ -8,30 +8,24 @@ from app.models import (
Assignment,
AssignmentSource,
Availability,
AvailabilityStatus,
DayPart,
Edition,
Position,
PositionCategory,
Shift,
Skill,
Volunteer,
)
EXPECTED_TABLES = {
"edition",
"day_part",
"skill",
"artist",
"volunteer",
"volunteer_skill",
"volunteer_artist",
"volunteer_position_wish",
"account",
"magic_link_token",
"position",
"shift",
"shift_required_skill",
"availability",
"assignment",
}
@@ -65,8 +59,6 @@ def test_create_core_graph(session):
first_name="Alex",
last_name="Martin",
)
skill = Skill(edition_id=edition.id, code="SST", label="Secourisme")
volunteer.skills.append(skill)
position = Position(
edition_id=edition.id, name="Bar principal", category=PositionCategory.BAR
)
@@ -76,9 +68,6 @@ def test_create_core_graph(session):
session.refresh(volunteer)
session.refresh(position)
# Lien N-N bénévole <-> compétence.
assert [s.code for s in volunteer.skills] == ["SST"]
shift = Shift(
edition_id=edition.id,
position_id=position.id,
@@ -107,37 +96,29 @@ def test_create_core_graph(session):
assert volunteer.assignments[0].id == assignment.id
def test_enum_round_trip(session):
def test_availability_window_round_trip(session):
edition = _edition(session)
day_part = DayPart(
edition_id=edition.id,
code="EVENING",
label="Soirée",
start_time=dt.time(19, 0),
end_time=dt.time(6, 0),
order=2,
)
volunteer = Volunteer(
edition_id=edition.id, first_name="Sam", last_name="Nuit"
)
session.add_all([day_part, volunteer])
session.add(volunteer)
session.commit()
session.refresh(day_part)
session.refresh(volunteer)
avail = Availability(
edition_id=edition.id,
volunteer_id=volunteer.id,
date=dt.date(2026, 7, 9),
day_part_id=day_part.id,
status=AvailabilityStatus.IF_NEEDED,
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.status is AvailabilityStatus.IF_NEEDED
assert reloaded.start_time == dt.time(19, 0)
assert reloaded.end_time == dt.time(2, 0)
def test_volunteer_email_unique_per_edition(session):
+50 -96
View File
@@ -3,9 +3,8 @@ règle, plus les cas limites de la sémantique figée (CLAUDE.md § 3)."""
import datetime as dt
from app.models import AvailabilityStatus
from app.validation import (
AvailabilityDeclaration,
AvailabilityWindow,
DayPartWindow,
Severity,
ShiftInfo,
@@ -26,18 +25,20 @@ def _dt(day: int, hour: int, minute: int = 0) -> dt.datetime:
return dt.datetime(2026, 7, day, hour, minute)
def _shift(sid: str, start: dt.datetime, end: dt.datetime, skills=()) -> ShiftInfo:
return ShiftInfo(
id=sid, starts_at=start, ends_at=end, required_skill_codes=frozenset(skills)
)
def _shift(sid: str, start: dt.datetime, end: dt.datetime) -> ShiftInfo:
return ShiftInfo(id=sid, starts_at=start, ends_at=end)
def _codes(violations) -> list[ViolationCode]:
return [v.code for v in violations]
def _avail(day: int, code: str, status: AvailabilityStatus) -> AvailabilityDeclaration:
return AvailabilityDeclaration(dt.date(2026, 7, day), code, status)
def _win(day: int, start_h: int, end_h: int) -> AvailabilityWindow:
"""Fenêtre de dispo le <day> juillet, de start_h:00 à end_h:00.
`end_h <= start_h` ⇒ se termine le lendemain."""
return AvailabilityWindow(
dt.date(2026, 7, day), dt.time(start_h % 24, 0), dt.time(end_h % 24, 0)
)
# --------------------------------------------------------------------------- #
@@ -47,10 +48,7 @@ def test_overlap_pass_when_shifts_are_adjacent():
data = ValidationInput(
candidate_shift=_shift("c", _dt(10, 14), _dt(10, 18)),
existing_shifts=(_shift("e", _dt(10, 18), _dt(10, 22)),),
day_parts=DAY_PARTS,
availabilities=(
_avail(10, "AFTERNOON", AvailabilityStatus.AVAILABLE),
),
availabilities=(_win(10, 8, 23),),
)
assert ViolationCode.SHIFT_OVERLAP not in _codes(validate(data))
@@ -59,6 +57,7 @@ def test_overlap_triggers_and_is_blocking():
data = ValidationInput(
candidate_shift=_shift("c", _dt(10, 14), _dt(10, 18)),
existing_shifts=(_shift("e", _dt(10, 16), _dt(10, 20)),),
availabilities=(_win(10, 8, 23),),
)
violations = validate(data)
overlap = [v for v in violations if v.code is ViolationCode.SHIFT_OVERLAP]
@@ -71,73 +70,68 @@ def test_overlap_triggers_and_is_blocking():
# --------------------------------------------------------------------------- #
# OUTSIDE_AVAILABILITY #
# --------------------------------------------------------------------------- #
def test_availability_pass_when_declared_available():
def test_availability_pass_when_shift_within_declared_window():
data = ValidationInput(
candidate_shift=_shift("c", _dt(10, 14), _dt(10, 18)),
day_parts=DAY_PARTS,
availabilities=(_avail(10, "AFTERNOON", AvailabilityStatus.AVAILABLE),),
availabilities=(_win(10, 13, 19),),
)
assert validate(data) == []
def test_availability_pass_when_declared_if_needed():
def test_availability_pass_when_window_exactly_matches_shift():
data = ValidationInput(
candidate_shift=_shift("c", _dt(10, 14), _dt(10, 18)),
day_parts=DAY_PARTS,
availabilities=(_avail(10, "AFTERNOON", AvailabilityStatus.IF_NEEDED),),
availabilities=(_win(10, 14, 18),),
)
assert validate(data) == []
def test_availability_triggers_when_slot_unavailable():
def test_availability_triggers_when_nothing_declared():
data = ValidationInput(
candidate_shift=_shift("c", _dt(10, 14), _dt(10, 18)),
day_parts=DAY_PARTS,
availabilities=(_avail(10, "AFTERNOON", AvailabilityStatus.UNAVAILABLE),),
)
violations = validate(data)
assert _codes(violations) == [ViolationCode.OUTSIDE_AVAILABILITY]
assert violations[0].severity is Severity.WARNING
assert violations[0].params["status"] == "UNAVAILABLE"
assert violations[0].params["day_part_code"] == "AFTERNOON"
assert violations[0].params["date"] == "2026-07-10"
assert violations[0].params["uncovered_minutes"] == 240
def test_availability_triggers_when_slot_undeclared():
def test_availability_triggers_on_partial_coverage_and_counts_the_gap():
# Créneau 16:00 -> 20:00 ; dispo seulement 13:00 -> 19:00 ⇒ 60 min à découvert.
data = ValidationInput(
candidate_shift=_shift("c", _dt(10, 14), _dt(10, 18)),
day_parts=DAY_PARTS,
candidate_shift=_shift("c", _dt(10, 16), _dt(10, 20)),
availabilities=(_win(10, 13, 19),),
)
violations = validate(data)
assert _codes(violations) == [ViolationCode.OUTSIDE_AVAILABILITY]
assert violations[0].params["status"] == "UNDECLARED"
assert violations[0].params["uncovered_minutes"] == 60
def test_availability_covers_two_slots_and_flags_only_the_missing_one():
# 18:00 -> 22:00 recoupe AFTERNOON (jusqu'à 19:00) et EVENING (dès 19:00).
def test_availability_union_of_several_windows_covers_the_shift():
data = ValidationInput(
candidate_shift=_shift("c", _dt(10, 18), _dt(10, 22)),
day_parts=DAY_PARTS,
availabilities=(_avail(10, "AFTERNOON", AvailabilityStatus.AVAILABLE),),
)
violations = validate(data)
assert len(violations) == 1
assert violations[0].params["day_part_code"] == "EVENING"
def test_availability_midnight_crossing_shift_uses_evening_slot_only():
data = ValidationInput(
candidate_shift=_shift("c", _dt(10, 22), _dt(11, 2)),
day_parts=DAY_PARTS,
availabilities=(_avail(10, "EVENING", AvailabilityStatus.AVAILABLE),),
candidate_shift=_shift("c", _dt(10, 12), _dt(10, 20)),
availabilities=(_win(10, 12, 16), _win(10, 15, 20)),
)
assert validate(data) == []
def test_availability_rule_skipped_without_day_parts():
def test_availability_midnight_crossing_window_covers_night_shift():
data = ValidationInput(
candidate_shift=_shift("c", _dt(10, 14), _dt(10, 18)),
candidate_shift=_shift("c", _dt(10, 22), _dt(11, 2)),
availabilities=(_win(10, 19, 4),), # 19:00 -> 04:00 le lendemain
)
assert ViolationCode.OUTSIDE_AVAILABILITY not in _codes(validate(data))
assert validate(data) == []
def test_availability_gap_between_two_windows_is_flagged():
data = ValidationInput(
candidate_shift=_shift("c", _dt(10, 10), _dt(10, 18)),
availabilities=(_win(10, 10, 12), _win(10, 14, 18)), # trou 12:00-14:00
)
violations = validate(data)
assert _codes(violations) == [ViolationCode.OUTSIDE_AVAILABILITY]
assert violations[0].params["uncovered_minutes"] == 120
# --------------------------------------------------------------------------- #
@@ -147,10 +141,7 @@ def test_daily_hours_pass_at_exactly_eight():
data = ValidationInput(
candidate_shift=_shift("c", _dt(10, 9), _dt(10, 17)), # 8 h pile
day_parts=DAY_PARTS,
availabilities=(
_avail(10, "MORNING", AvailabilityStatus.AVAILABLE),
_avail(10, "AFTERNOON", AvailabilityStatus.AVAILABLE),
),
availabilities=(_win(10, 8, 18),),
)
assert ViolationCode.DAILY_HOURS_EXCEEDED not in _codes(validate(data))
@@ -200,8 +191,7 @@ def test_rest_pass_with_more_than_eight_hours_gap():
data = ValidationInput(
candidate_shift=_shift("c", _dt(10, 8), _dt(10, 12)),
existing_shifts=(_shift("e", _dt(9, 18), _dt(9, 23)),), # écart 9 h
day_parts=DAY_PARTS,
availabilities=(_avail(10, "MORNING", AvailabilityStatus.AVAILABLE),),
availabilities=(_win(10, 8, 12),),
)
assert ViolationCode.INSUFFICIENT_REST not in _codes(validate(data))
@@ -243,81 +233,45 @@ def test_back_to_back_shifts_are_rest_violation_not_overlap():
data = ValidationInput(
candidate_shift=_shift("c", _dt(10, 14), _dt(10, 18)),
existing_shifts=(_shift("e", _dt(10, 18), _dt(10, 22)),),
availabilities=(_win(10, 8, 23),),
)
codes = _codes(validate(data))
assert ViolationCode.SHIFT_OVERLAP not in codes
assert ViolationCode.INSUFFICIENT_REST in codes
# --------------------------------------------------------------------------- #
# MISSING_REQUIRED_SKILL #
# --------------------------------------------------------------------------- #
def test_skill_pass_when_volunteer_has_all_required():
data = ValidationInput(
candidate_shift=_shift("c", _dt(10, 14), _dt(10, 18), skills=["SST"]),
volunteer_skill_codes=frozenset({"SST", "PERMIS_B"}),
day_parts=DAY_PARTS,
availabilities=(_avail(10, "AFTERNOON", AvailabilityStatus.AVAILABLE),),
)
assert validate(data) == []
def test_skill_triggers_for_each_missing_code_sorted():
data = ValidationInput(
candidate_shift=_shift(
"c", _dt(10, 14), _dt(10, 18), skills=["SOUND", "SST", "PERMIS_B"]
),
volunteer_skill_codes=frozenset({"SST"}),
day_parts=DAY_PARTS,
availabilities=(_avail(10, "AFTERNOON", AvailabilityStatus.AVAILABLE),),
)
violations = validate(data)
assert [v.params["skill_code"] for v in violations] == ["PERMIS_B", "SOUND"]
assert all(v.severity is Severity.WARNING for v in violations)
# --------------------------------------------------------------------------- #
# Combinaisons #
# --------------------------------------------------------------------------- #
def test_clean_assignment_has_no_violation():
data = ValidationInput(
candidate_shift=_shift("c", _dt(10, 14), _dt(10, 18), skills=["SST"]),
volunteer_skill_codes=frozenset({"SST"}),
candidate_shift=_shift("c", _dt(10, 14), _dt(10, 18)),
existing_shifts=(_shift("e", _dt(9, 8), _dt(9, 12)),),
day_parts=DAY_PARTS,
availabilities=(_avail(10, "AFTERNOON", AvailabilityStatus.AVAILABLE),),
availabilities=(_win(10, 13, 19),),
)
assert validate(data) == []
def test_multiple_rules_fire_together():
data = ValidationInput(
candidate_shift=_shift(
"c", _dt(10, 14), _dt(10, 18), skills=["SOUND"]
),
volunteer_skill_codes=frozenset(),
candidate_shift=_shift("c", _dt(10, 14), _dt(10, 18)),
existing_shifts=(_shift("e", _dt(10, 16), _dt(10, 20)),), # chevauchement
day_parts=DAY_PARTS,
availabilities=(
_avail(10, "AFTERNOON", AvailabilityStatus.UNAVAILABLE),
),
# aucune dispo déclarée ⇒ OUTSIDE_AVAILABILITY
)
codes = set(_codes(validate(data)))
assert {
ViolationCode.SHIFT_OVERLAP,
ViolationCode.OUTSIDE_AVAILABILITY,
ViolationCode.MISSING_REQUIRED_SKILL,
} <= codes
assert has_blocking(validate(data)) is True
def test_only_warnings_is_not_blocking():
data = ValidationInput(
candidate_shift=_shift("c", _dt(10, 14), _dt(10, 18), skills=["SOUND"]),
volunteer_skill_codes=frozenset(),
day_parts=DAY_PARTS,
availabilities=(_avail(10, "AFTERNOON", AvailabilityStatus.AVAILABLE),),
candidate_shift=_shift("c", _dt(10, 14), _dt(10, 18)),
# pas de dispo ⇒ un seul avertissement OUTSIDE_AVAILABILITY
)
violations = validate(data)
assert _codes(violations) == [ViolationCode.MISSING_REQUIRED_SKILL]
assert _codes(violations) == [ViolationCode.OUTSIDE_AVAILABILITY]
assert has_blocking(violations) is False
+8 -14
View File
@@ -1,7 +1,7 @@
import datetime as dt
from app.models import Assignment, AvailabilityStatus
from app.validation import AvailabilityDeclaration
from app.models import Assignment
from app.validation import AvailabilityWindow
from app.validation.loader import build_validation_input
from tests.factories import (
dt_at,
@@ -15,18 +15,16 @@ from tests.factories import (
def test_build_input_gathers_everything_from_db(client, edition, session, day_parts):
position = make_position(session, edition)
candidate = make_shift(
session, edition, position, start=dt_at(10, 14), end=dt_at(10, 18),
skills=["SST"],
session, edition, position, start=dt_at(10, 14), end=dt_at(10, 18)
)
other_shift = make_shift(
session, edition, position, start=dt_at(11, 9), end=dt_at(11, 12)
)
volunteer = make_volunteer(
session, edition, email="v@ex.fr", skills=["SST", "PERMIS_B"]
)
volunteer = make_volunteer(session, edition, email="v@ex.fr")
make_availability(
session, edition, volunteer, day_parts["AFTERNOON"],
date=dt.date(2026, 7, 10), status=AvailabilityStatus.IF_NEEDED,
session, edition, volunteer,
date=dt.date(2026, 7, 10),
start_time=dt.time(13, 0), end_time=dt.time(19, 0),
)
session.add(
Assignment(
@@ -41,15 +39,11 @@ def test_build_input_gathers_everything_from_db(client, edition, session, day_pa
)
assert data.candidate_shift.id == str(candidate.id)
assert data.candidate_shift.required_skill_codes == frozenset({"SST"})
assert data.volunteer_skill_codes == frozenset({"SST", "PERMIS_B"})
# Toute affectation existante compte, sauf le créneau candidat lui-même.
assert [s.id for s in data.existing_shifts] == [str(other_shift.id)]
assert len(data.day_parts) == 3
assert data.availabilities == (
AvailabilityDeclaration(
dt.date(2026, 7, 10), "AFTERNOON", AvailabilityStatus.IF_NEEDED
),
AvailabilityWindow(dt.date(2026, 7, 10), dt.time(13, 0), dt.time(19, 0)),
)