Étape 5 : API CRUD, matrice et validation branchée
- CRUD complet, scopé par édition, sur Position, Shift, Volunteer, Assignment
(routers app/api/*, schémas Pydantic, dépendance get_edition partagée)
- Volunteer : gestion des compétences par codes (création à la volée),
suppression = effacement complet en cascade (RGPD)
- Shift : contrôle fin > début, poste de la même édition, compétences requises
- Assignment : toute écriture active passe par le moteur de validation ;
violation bloquante -> 422 error.validation_failed, avertissements renvoyés
dans la réponse. Endpoint POST .../assignments/validate en simulation.
app/validation/loader.py construit la ValidationInput depuis la base.
- GET /editions/{id}/matrix : postes -> créneaux -> {min, ideal, confirmed,
proposed}, source de l'écran principal
- 30 tests d'API (CRUD, 404/409/422, cascade, blocage, avertissements, matrice,
loader) ; 104 tests au total ; alembic check toujours vert
- Clés de traduction shift.* / assignment.* / volunteer.* ajoutées
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017uUsRg7XzopkCPWimwg9ZM
This commit is contained in:
AdamThomas
co-authored by
Claude Sonnet 5
Claude Sonnet 5
parent
c96e85af37
commit
74e35f0195
@@ -2,7 +2,7 @@ import datetime as dt
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlmodel import Session, SQLModel, create_engine
|
||||
from sqlmodel import Session, SQLModel, create_engine, select
|
||||
|
||||
from app import models # noqa: F401 (enregistre les tables sur SQLModel.metadata)
|
||||
from app.config import get_settings
|
||||
@@ -75,3 +75,11 @@ def edition(session) -> Edition:
|
||||
)
|
||||
session.commit()
|
||||
return ed
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def day_parts(session, edition) -> dict[str, DayPart]:
|
||||
parts = session.exec(
|
||||
select(DayPart).where(DayPart.edition_id == edition.id)
|
||||
)
|
||||
return {p.code: p for p in parts}
|
||||
|
||||
@@ -0,0 +1,118 @@
|
||||
"""Fabriques d'objets du domaine pour les tests (écriture directe en base)."""
|
||||
|
||||
import datetime as dt
|
||||
|
||||
from app.models import (
|
||||
Availability,
|
||||
AvailabilityStatus,
|
||||
Position,
|
||||
PositionCategory,
|
||||
Shift,
|
||||
Skill,
|
||||
Volunteer,
|
||||
VolunteerSkill,
|
||||
)
|
||||
|
||||
|
||||
def make_position(session, edition, *, name="Bar", category=PositionCategory.BAR):
|
||||
position = Position(edition_id=edition.id, name=name, category=category)
|
||||
session.add(position)
|
||||
session.commit()
|
||||
session.refresh(position)
|
||||
return position
|
||||
|
||||
|
||||
def make_shift(
|
||||
session, edition, position, *, start, end, min_staff=1, ideal_staff=2, skills=()
|
||||
):
|
||||
shift = Shift(
|
||||
edition_id=edition.id,
|
||||
position_id=position.id,
|
||||
starts_at=start,
|
||||
ends_at=end,
|
||||
min_staff=min_staff,
|
||||
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=()
|
||||
):
|
||||
volunteer = Volunteer(
|
||||
edition_id=edition.id,
|
||||
first_name=first_name,
|
||||
last_name=last_name,
|
||||
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):
|
||||
row = Availability(
|
||||
edition_id=edition.id,
|
||||
volunteer_id=volunteer.id,
|
||||
day_part_id=day_part.id,
|
||||
date=date,
|
||||
status=status,
|
||||
)
|
||||
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 dt_at(day: int, hour: int, minute: int = 0) -> dt.datetime:
|
||||
return dt.datetime(2026, 7, day, hour, minute)
|
||||
@@ -0,0 +1,195 @@
|
||||
import datetime as dt
|
||||
|
||||
from tests.factories import (
|
||||
declare_available_all_day,
|
||||
dt_at,
|
||||
make_position,
|
||||
make_shift,
|
||||
make_volunteer,
|
||||
)
|
||||
|
||||
|
||||
def _url(edition, suffix=""):
|
||||
return f"/editions/{edition.id}/assignments{suffix}"
|
||||
|
||||
|
||||
def _setup(session, edition, day_parts, *, available=True, skills=()):
|
||||
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)
|
||||
if available:
|
||||
declare_available_all_day(
|
||||
session, edition, volunteer, day_parts, 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)
|
||||
resp = client.post(
|
||||
_url(edition),
|
||||
json={"volunteer_id": str(volunteer.id), "shift_id": str(shift.id)},
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
body = resp.json()
|
||||
assert body["warnings"] == []
|
||||
assert body["assignment"]["status"] == "PROPOSED"
|
||||
assert body["assignment"]["source"] == "MANUAL"
|
||||
|
||||
|
||||
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"],
|
||||
)
|
||||
resp = client.post(
|
||||
_url(edition),
|
||||
json={
|
||||
"volunteer_id": str(volunteer.id),
|
||||
"shift_id": str(shift_with_skill.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)
|
||||
assert client.post(
|
||||
_url(edition),
|
||||
json={"volunteer_id": str(volunteer.id), "shift_id": str(shift.id)},
|
||||
).status_code == 201
|
||||
|
||||
position = make_position(session, edition, name="Parking")
|
||||
overlapping = make_shift(
|
||||
session, edition, position, start=dt_at(10, 16), end=dt_at(10, 20)
|
||||
)
|
||||
resp = client.post(
|
||||
_url(edition),
|
||||
json={
|
||||
"volunteer_id": str(volunteer.id),
|
||||
"shift_id": str(overlapping.id),
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
detail = resp.json()["detail"]
|
||||
assert detail["code"] == "error.validation_failed"
|
||||
assert any(v["code"] == "SHIFT_OVERLAP" for v in detail["violations"])
|
||||
# Rien n'a été enregistré pour ce créneau.
|
||||
listed = client.get(
|
||||
_url(edition), params={"shift_id": str(overlapping.id)}
|
||||
).json()
|
||||
assert listed == []
|
||||
|
||||
|
||||
def test_create_duplicate_assignment_conflicts(client, edition, session, day_parts):
|
||||
volunteer, shift = _setup(session, edition, day_parts)
|
||||
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)
|
||||
assert resp.status_code == 409
|
||||
assert resp.json()["detail"]["code"] == "assignment.duplicate"
|
||||
|
||||
|
||||
def test_assignment_wrong_edition_is_422(client, edition, session, day_parts):
|
||||
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="s@ex.fr")
|
||||
_, shift = _setup(session, edition, day_parts)
|
||||
|
||||
resp = client.post(
|
||||
_url(edition),
|
||||
json={"volunteer_id": str(stranger.id), "shift_id": str(shift.id)},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
assert resp.json()["detail"]["code"] == "assignment.wrong_edition"
|
||||
|
||||
|
||||
def test_validate_endpoint_does_not_persist(client, edition, session, day_parts):
|
||||
volunteer, shift = _setup(session, edition, day_parts, available=False)
|
||||
resp = client.post(
|
||||
_url(edition, "/validate"),
|
||||
json={"volunteer_id": str(volunteer.id), "shift_id": str(shift.id)},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert body["blocking"] is False
|
||||
assert any(v["code"] == "OUTSIDE_AVAILABILITY" for v in body["violations"])
|
||||
assert client.get(_url(edition)).json() == []
|
||||
|
||||
|
||||
def test_update_status_revalidates_and_returns_warnings(
|
||||
client, edition, session, day_parts
|
||||
):
|
||||
volunteer, shift = _setup(session, edition, day_parts, available=False)
|
||||
aid = client.post(
|
||||
_url(edition),
|
||||
json={"volunteer_id": str(volunteer.id), "shift_id": str(shift.id)},
|
||||
).json()["assignment"]["id"]
|
||||
|
||||
resp = client.patch(
|
||||
_url(edition, f"/{aid}"), json={"status": "CONFIRMED"}
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["assignment"]["status"] == "CONFIRMED"
|
||||
assert any(
|
||||
w["code"] == "OUTSIDE_AVAILABILITY" for w in resp.json()["warnings"]
|
||||
)
|
||||
|
||||
|
||||
def test_cancelling_an_assignment_skips_validation(
|
||||
client, edition, session, day_parts
|
||||
):
|
||||
volunteer, shift = _setup(session, edition, day_parts)
|
||||
first = client.post(
|
||||
_url(edition),
|
||||
json={"volunteer_id": str(volunteer.id), "shift_id": str(shift.id)},
|
||||
).json()["assignment"]["id"]
|
||||
|
||||
# Un second créneau qui chevauche, mais l'affectation est directement DECLINEE.
|
||||
position = make_position(session, edition, name="X")
|
||||
overlapping = make_shift(
|
||||
session, edition, position, start=dt_at(10, 16), end=dt_at(10, 20)
|
||||
)
|
||||
created = client.post(
|
||||
_url(edition),
|
||||
json={
|
||||
"volunteer_id": str(volunteer.id),
|
||||
"shift_id": str(overlapping.id),
|
||||
"status": "DECLINED",
|
||||
},
|
||||
)
|
||||
assert created.status_code == 201
|
||||
|
||||
patched = client.patch(
|
||||
_url(edition, f"/{first}"), json={"locked": True}
|
||||
)
|
||||
assert patched.status_code == 200
|
||||
assert patched.json()["assignment"]["locked"] is True
|
||||
|
||||
|
||||
def test_delete_assignment(client, edition, session, day_parts):
|
||||
volunteer, shift = _setup(session, edition, day_parts)
|
||||
aid = client.post(
|
||||
_url(edition),
|
||||
json={"volunteer_id": str(volunteer.id), "shift_id": str(shift.id)},
|
||||
).json()["assignment"]["id"]
|
||||
assert client.delete(_url(edition, f"/{aid}")).status_code == 204
|
||||
assert client.get(_url(edition, f"/{aid}")).status_code == 404
|
||||
@@ -0,0 +1,62 @@
|
||||
|
||||
from app.models import Assignment, AssignmentStatus
|
||||
from tests.factories import dt_at, make_position, make_shift, make_volunteer
|
||||
|
||||
|
||||
def test_matrix_shape_and_counts(client, edition, session):
|
||||
bar = make_position(session, edition, name="Bar")
|
||||
parking = make_position(session, edition, name="Parking")
|
||||
|
||||
shift_a = make_shift(
|
||||
session, edition, bar, start=dt_at(10, 18), end=dt_at(10, 22),
|
||||
min_staff=2, ideal_staff=5,
|
||||
)
|
||||
shift_b = make_shift(
|
||||
session, edition, bar, start=dt_at(11, 18), end=dt_at(11, 22)
|
||||
)
|
||||
make_shift(
|
||||
session, edition, parking, start=dt_at(10, 8), end=dt_at(10, 12)
|
||||
)
|
||||
|
||||
for i, status in enumerate(
|
||||
[
|
||||
AssignmentStatus.CONFIRMED,
|
||||
AssignmentStatus.CONFIRMED,
|
||||
AssignmentStatus.PROPOSED,
|
||||
AssignmentStatus.DECLINED,
|
||||
]
|
||||
):
|
||||
volunteer = make_volunteer(
|
||||
session, edition, first_name=f"V{i}", last_name="X", email=f"v{i}@ex.fr"
|
||||
)
|
||||
session.add(
|
||||
Assignment(
|
||||
edition_id=edition.id,
|
||||
volunteer_id=volunteer.id,
|
||||
shift_id=shift_a.id,
|
||||
status=status,
|
||||
)
|
||||
)
|
||||
session.commit()
|
||||
|
||||
body = client.get(f"/editions/{edition.id}/matrix").json()
|
||||
assert body["edition_id"] == str(edition.id)
|
||||
assert [p["name"] for p in body["positions"]] == ["Bar", "Parking"]
|
||||
|
||||
bar_shifts = body["positions"][0]["shifts"]
|
||||
assert [s["shift_id"] for s in bar_shifts] == [str(shift_a.id), str(shift_b.id)]
|
||||
|
||||
first = bar_shifts[0]
|
||||
assert first["confirmed"] == 2
|
||||
assert first["proposed"] == 1 # la DECLINEE n'est pas comptée
|
||||
assert first["min_staff"] == 2
|
||||
assert first["ideal_staff"] == 5
|
||||
|
||||
assert bar_shifts[1]["confirmed"] == 0
|
||||
assert body["positions"][1]["shifts"][0]["proposed"] == 0
|
||||
|
||||
|
||||
def test_matrix_unknown_edition_is_404(client):
|
||||
import uuid
|
||||
|
||||
assert client.get(f"/editions/{uuid.uuid4()}/matrix").status_code == 404
|
||||
@@ -0,0 +1,37 @@
|
||||
import uuid
|
||||
|
||||
|
||||
def _url(edition, suffix=""):
|
||||
return f"/editions/{edition.id}/positions{suffix}"
|
||||
|
||||
|
||||
def test_position_crud_roundtrip(client, edition):
|
||||
created = client.post(
|
||||
_url(edition),
|
||||
json={"name": "Billetterie", "category": "TICKETING", "location": "Entrée A"},
|
||||
)
|
||||
assert created.status_code == 201
|
||||
pid = created.json()["id"]
|
||||
assert created.json()["category"] == "TICKETING"
|
||||
|
||||
assert client.get(_url(edition, f"/{pid}")).json()["name"] == "Billetterie"
|
||||
assert [p["name"] for p in client.get(_url(edition)).json()] == ["Billetterie"]
|
||||
|
||||
patched = client.patch(
|
||||
_url(edition, f"/{pid}"), json={"location": "Entrée B"}
|
||||
)
|
||||
assert patched.json()["location"] == "Entrée B"
|
||||
assert patched.json()["name"] == "Billetterie"
|
||||
|
||||
assert client.delete(_url(edition, f"/{pid}")).status_code == 204
|
||||
assert client.get(_url(edition, f"/{pid}")).status_code == 404
|
||||
|
||||
|
||||
def test_unknown_position_is_404(client, edition):
|
||||
assert client.get(_url(edition, f"/{uuid.uuid4()}")).status_code == 404
|
||||
|
||||
|
||||
def test_unknown_edition_is_404(client):
|
||||
assert (
|
||||
client.get(f"/editions/{uuid.uuid4()}/positions").status_code == 404
|
||||
)
|
||||
@@ -0,0 +1,100 @@
|
||||
import datetime as dt
|
||||
|
||||
from tests.factories import make_position
|
||||
|
||||
|
||||
def _url(edition, suffix=""):
|
||||
return f"/editions/{edition.id}/shifts{suffix}"
|
||||
|
||||
|
||||
def _payload(position, **overrides):
|
||||
base = {
|
||||
"position_id": str(position.id),
|
||||
"starts_at": "2026-07-10T18:00:00",
|
||||
"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):
|
||||
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)
|
||||
|
||||
|
||||
def test_create_shift_end_before_start_is_422(client, edition, session):
|
||||
position = make_position(session, edition)
|
||||
resp = client.post(
|
||||
_url(edition),
|
||||
json=_payload(position, ends_at="2026-07-10T17:00:00"),
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
assert resp.json()["detail"]["code"] == "shift.ends_not_after_starts"
|
||||
|
||||
|
||||
def test_create_shift_with_position_from_other_edition_is_422(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_position = make_position(session, other)
|
||||
|
||||
resp = client.post(_url(edition), json=_payload(stranger_position))
|
||||
assert resp.status_code == 422
|
||||
assert resp.json()["detail"]["code"] == "shift.position_wrong_edition"
|
||||
|
||||
|
||||
def test_list_filter_by_position(client, edition, session):
|
||||
p1 = make_position(session, edition, name="P1")
|
||||
p2 = make_position(session, edition, name="P2")
|
||||
client.post(_url(edition), json=_payload(p1))
|
||||
client.post(
|
||||
_url(edition),
|
||||
json=_payload(
|
||||
p2,
|
||||
starts_at="2026-07-11T09:00:00",
|
||||
ends_at="2026-07-11T12:00:00",
|
||||
),
|
||||
)
|
||||
|
||||
only_p1 = client.get(_url(edition), params={"position_id": str(p1.id)}).json()
|
||||
assert len(only_p1) == 1
|
||||
assert only_p1[0]["position_id"] == str(p1.id)
|
||||
assert len(client.get(_url(edition)).json()) == 2
|
||||
|
||||
|
||||
def test_patch_shift_times_and_skills(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"],
|
||||
},
|
||||
)
|
||||
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):
|
||||
position = make_position(session, edition)
|
||||
sid = client.post(_url(edition), json=_payload(position)).json()["id"]
|
||||
resp = client.patch(
|
||||
_url(edition, f"/{sid}"), json={"starts_at": "2026-07-11T00:00:00"}
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
@@ -0,0 +1,130 @@
|
||||
import datetime as dt
|
||||
|
||||
from sqlmodel import select
|
||||
|
||||
from app.models import Assignment, Availability, AvailabilityStatus, Volunteer
|
||||
from tests.factories import (
|
||||
dt_at,
|
||||
make_availability,
|
||||
make_position,
|
||||
make_shift,
|
||||
make_volunteer,
|
||||
)
|
||||
|
||||
|
||||
def _url(edition, suffix=""):
|
||||
return f"/editions/{edition.id}/volunteers{suffix}"
|
||||
|
||||
|
||||
def test_create_volunteer_with_skills(client, edition):
|
||||
resp = client.post(
|
||||
_url(edition),
|
||||
json={
|
||||
"first_name": "Alex",
|
||||
"last_name": "Martin",
|
||||
"email": "alex@ex.fr",
|
||||
"tshirt_size": "M",
|
||||
"skill_codes": ["SST", "PERMIS_B", "SST"],
|
||||
},
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
body = resp.json()
|
||||
assert body["skill_codes"] == ["PERMIS_B", "SST"]
|
||||
assert body["tshirt_size"] == "M"
|
||||
assert body["edition_id"] == str(edition.id)
|
||||
|
||||
|
||||
def test_create_duplicate_email_conflicts(client, edition):
|
||||
payload = {"first_name": "A", "last_name": "A", "email": "dup@ex.fr"}
|
||||
assert client.post(_url(edition), json=payload).status_code == 201
|
||||
resp = client.post(_url(edition), json={**payload, "first_name": "B"})
|
||||
assert resp.status_code == 409
|
||||
assert resp.json()["detail"]["code"] == "volunteer.duplicate_email"
|
||||
|
||||
|
||||
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):
|
||||
import uuid
|
||||
|
||||
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_patch_replaces_skills_and_can_clear(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"]},
|
||||
)
|
||||
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"] == []
|
||||
|
||||
|
||||
def test_delete_is_a_full_cascade(client, edition, session, day_parts):
|
||||
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,
|
||||
day_parts["MORNING"],
|
||||
date=dt.date(2026, 7, 10),
|
||||
status=AvailabilityStatus.AVAILABLE,
|
||||
)
|
||||
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
|
||||
@@ -0,0 +1,86 @@
|
||||
import datetime as dt
|
||||
|
||||
from app.models import Assignment, AssignmentStatus, AvailabilityStatus
|
||||
from app.validation import AvailabilityDeclaration
|
||||
from app.validation.loader import build_validation_input
|
||||
from tests.factories import (
|
||||
dt_at,
|
||||
make_availability,
|
||||
make_position,
|
||||
make_shift,
|
||||
make_volunteer,
|
||||
)
|
||||
|
||||
|
||||
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"],
|
||||
)
|
||||
active_shift = make_shift(
|
||||
session, edition, position, start=dt_at(11, 9), end=dt_at(11, 12)
|
||||
)
|
||||
declined_shift = make_shift(
|
||||
session, edition, position, start=dt_at(12, 9), end=dt_at(12, 12)
|
||||
)
|
||||
volunteer = make_volunteer(
|
||||
session, edition, email="v@ex.fr", skills=["SST", "PERMIS_B"]
|
||||
)
|
||||
make_availability(
|
||||
session, edition, volunteer, day_parts["AFTERNOON"],
|
||||
date=dt.date(2026, 7, 10), status=AvailabilityStatus.IF_NEEDED,
|
||||
)
|
||||
session.add_all(
|
||||
[
|
||||
Assignment(
|
||||
edition_id=edition.id, volunteer_id=volunteer.id,
|
||||
shift_id=active_shift.id, status=AssignmentStatus.CONFIRMED,
|
||||
),
|
||||
Assignment(
|
||||
edition_id=edition.id, volunteer_id=volunteer.id,
|
||||
shift_id=declined_shift.id, status=AssignmentStatus.DECLINED,
|
||||
),
|
||||
]
|
||||
)
|
||||
session.commit()
|
||||
|
||||
data = build_validation_input(
|
||||
session=session, edition=edition, volunteer=volunteer, shift=candidate
|
||||
)
|
||||
|
||||
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"})
|
||||
# Seul le créneau de l'affectation active, ni le candidat ni la DECLINEE.
|
||||
assert [s.id for s in data.existing_shifts] == [str(active_shift.id)]
|
||||
assert len(data.day_parts) == 3
|
||||
assert data.availabilities == (
|
||||
AvailabilityDeclaration(
|
||||
dt.date(2026, 7, 10), "AFTERNOON", AvailabilityStatus.IF_NEEDED
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def test_exclude_assignment_id_is_honoured(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)
|
||||
)
|
||||
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")
|
||||
own = Assignment(
|
||||
edition_id=edition.id, volunteer_id=volunteer.id,
|
||||
shift_id=other_shift.id, status=AssignmentStatus.PROPOSED,
|
||||
)
|
||||
session.add(own)
|
||||
session.commit()
|
||||
session.refresh(own)
|
||||
|
||||
data = build_validation_input(
|
||||
session=session, edition=edition, volunteer=volunteer,
|
||||
shift=candidate, exclude_assignment_id=own.id,
|
||||
)
|
||||
assert data.existing_shifts == ()
|
||||
Reference in New Issue
Block a user