Étape 2 : modèle de données complet et migration initiale
- Tables SQLModel : Edition, DayPart, Volunteer, Skill, Position, Shift, Availability, Assignment, Account, MagicLinkToken + tables d'association - Multi-édition (edition_id partout), énumérations en VARCHAR + CHECK, suppressions ON DELETE CASCADE (droit à l'effacement) - Dates/heures naïves en heure locale Europe/Paris (tzdata + TZ conteneur) - Migration Alembic 0001 écrite à la main, vérifiée par `alembic check` (aucun écart avec les modèles) et par un aller-retour downgrade/upgrade - Tests : graphe complet, unicité email par édition, N-N compétences, aller-retour d'énumération, cascade de suppression d'un bénévole 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
09d7d70076
commit
cb2e84adf0
@@ -12,6 +12,7 @@ from app.main import app
|
||||
def engine():
|
||||
"""Vraie base PostgreSQL de test (voir docker/postgres-init)."""
|
||||
eng = create_engine(get_settings().test_database_url, pool_pre_ping=True)
|
||||
SQLModel.metadata.drop_all(eng)
|
||||
SQLModel.metadata.create_all(eng)
|
||||
yield eng
|
||||
SQLModel.metadata.drop_all(eng)
|
||||
@@ -19,13 +20,17 @@ def engine():
|
||||
|
||||
@pytest.fixture
|
||||
def session(engine):
|
||||
"""Session isolée : tout est annulé à la fin du test."""
|
||||
"""Session isolée : chaque test s'exécute dans une transaction annulée
|
||||
à la fin, y compris ses `commit()` (savepoints)."""
|
||||
connection = engine.connect()
|
||||
transaction = connection.begin()
|
||||
with Session(bind=connection) as s:
|
||||
s = Session(bind=connection, join_transaction_mode="create_savepoint")
|
||||
try:
|
||||
yield s
|
||||
transaction.rollback()
|
||||
connection.close()
|
||||
finally:
|
||||
s.close()
|
||||
transaction.rollback()
|
||||
connection.close()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
||||
@@ -0,0 +1,216 @@
|
||||
import datetime as dt
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
|
||||
from app.models import (
|
||||
Assignment,
|
||||
AssignmentSource,
|
||||
AssignmentStatus,
|
||||
Availability,
|
||||
AvailabilityStatus,
|
||||
DayPart,
|
||||
Edition,
|
||||
Position,
|
||||
PositionCategory,
|
||||
Shift,
|
||||
Skill,
|
||||
Volunteer,
|
||||
)
|
||||
|
||||
EXPECTED_TABLES = {
|
||||
"edition",
|
||||
"day_part",
|
||||
"skill",
|
||||
"volunteer",
|
||||
"volunteer_skill",
|
||||
"account",
|
||||
"magic_link_token",
|
||||
"position",
|
||||
"shift",
|
||||
"shift_required_skill",
|
||||
"availability",
|
||||
"assignment",
|
||||
}
|
||||
|
||||
|
||||
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",
|
||||
)
|
||||
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
|
||||
)
|
||||
session.add(volunteer)
|
||||
session.add(position)
|
||||
session.commit()
|
||||
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,
|
||||
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.status == AssignmentStatus.PROPOSED
|
||||
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_enum_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.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,
|
||||
)
|
||||
session.add(avail)
|
||||
session.commit()
|
||||
session.expire_all()
|
||||
|
||||
reloaded = session.get(Availability, avail.id)
|
||||
assert reloaded.status is AvailabilityStatus.IF_NEEDED
|
||||
|
||||
|
||||
def test_volunteer_email_unique_per_edition(session):
|
||||
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",
|
||||
)
|
||||
)
|
||||
with pytest.raises(IntegrityError):
|
||||
session.commit()
|
||||
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user