from sqlmodel import select from app.models import Assignment, MagicLinkToken, Role from tests.factories import dt_at, make_position, make_shift, make_volunteer from tests.mail import plain_text def _url(edition): return f"/editions/{edition.id}/notifications/planning" def test_one_email_per_volunteer_with_an_address(client, edition, session, outbox): position = make_position(session, edition, name="Bar") s1 = make_shift(session, edition, position, start=dt_at(10, 9), end=dt_at(10, 13)) s2 = make_shift(session, edition, position, start=dt_at(11, 9), end=dt_at(11, 12)) ana = make_volunteer(session, edition, first_name="Ana", email="ana@e.fr") make_volunteer(session, edition, first_name="Bob", email="bob@e.fr") make_volunteer(session, edition, first_name="Cid", email=None) session.add_all( [ Assignment(edition_id=edition.id, volunteer_id=ana.id, shift_id=s1.id), Assignment(edition_id=edition.id, volunteer_id=ana.id, shift_id=s2.id), ] ) session.commit() resp = client.post( _url(edition), json={ "subject": "Planning de {prenom}", "body": "Tu as {nb_creneaux} créneau(x), {nb_heures} h.\n{creneaux}", }, ) assert resp.status_code == 200 body = resp.json() assert body["queued"] == 2 assert body["skipped_no_email"] == 1 assert len(outbox) == 2 by_to = {m["To"]: m for m in outbox} assert by_to["ana@e.fr"]["Subject"] == "Planning de Ana" ana_body = plain_text(by_to["ana@e.fr"]) assert "Tu as 2 créneau(x), 7.0 h." in ana_body assert "Bar" in ana_body assert "Tu as 0 créneau(x), 0.0 h." in plain_text(by_to["bob@e.fr"]) def test_default_template_is_used_when_body_omitted(client, edition, session, outbox): make_volunteer(session, edition, first_name="Ana", email="ana@e.fr") resp = client.post(_url(edition), json={}) assert resp.status_code == 200 assert plain_text(outbox[0]).startswith("Bonjour Ana,") def test_magic_link_is_embedded_when_placeholder_present( client, edition, session, make_account, outbox ): ana = make_volunteer(session, edition, first_name="Ana", email="ana@e.fr") make_account(Role.VOLUNTEER, volunteer_id=ana.id, email="ana@e.fr") resp = client.post( _url(edition), json={"body": "Connecte-toi : {lien_connexion}"} ) assert resp.status_code == 200 assert "/connexion?token=pl." in plain_text(outbox[0]) # Lien permanent : rien de stocké, pas de jeton à usage unique. assert session.exec(select(MagicLinkToken)).all() == [] def test_no_magic_link_token_without_placeholder( client, edition, session, make_account, outbox ): ana = make_volunteer(session, edition, first_name="Ana", email="ana@e.fr") make_account(Role.VOLUNTEER, volunteer_id=ana.id, email="ana@e.fr") client.post(_url(edition), json={"body": "Bonjour {prenom}"}) assert session.exec(select(MagicLinkToken)).all() == [] def test_can_target_a_subset(client, edition, session, outbox): ana = make_volunteer(session, edition, email="ana@e.fr") make_volunteer(session, edition, email="bob@e.fr") resp = client.post( _url(edition), json={"volunteer_ids": [str(ana.id)]} ) assert resp.json()["queued"] == 1 assert outbox[0]["To"] == "ana@e.fr" def test_planning_notification_is_coordinator_only( edition, session, make_account, as_account, outbox ): lead = make_account(Role.LEAD, email="lead@e.fr") resp = as_account(lead).post(_url(edition), json={}) assert resp.status_code == 403 def test_shifts_placeholder_is_a_table_by_day_with_the_referent( client, edition, session, outbox, make_account ): rita = make_volunteer(session, edition, first_name="Rita", last_name="Ref", email="r@e.fr") rita.phone = "06 11 22 33 44" session.add(rita) lead = make_account(Role.LEAD, volunteer_id=rita.id) bar = make_position(session, edition, name="Bar central") bar.lead_account_id = lead.id session.add(bar) parking = make_position(session, edition, name="Parking") s1 = make_shift(session, edition, bar, start=dt_at(10, 18), end=dt_at(10, 22)) s2 = make_shift(session, edition, parking, start=dt_at(11, 9), end=dt_at(11, 12)) ana = make_volunteer(session, edition, first_name="Ana", email="ana@e.fr") session.add_all( [ Assignment(edition_id=edition.id, volunteer_id=ana.id, shift_id=s1.id), Assignment(edition_id=edition.id, volunteer_id=ana.id, shift_id=s2.id), ] ) session.commit() client.post( _url(edition), json={"volunteer_ids": [str(ana.id)], "body": "Ton planning :\n\n{creneaux}"} ) (message,) = outbox html = message.get_body(preferencelist=("html",)).get_content() # Vrai tableau : en-têtes, une ligne de jour, horaire, poste, référent. assert " str: import re return re.search(r"token=(\S+)", plain_text(message)).group(1) def _verify(client, edition, token): return client.post(f"/editions/{edition.id}/auth/verify", json={"token": token}) def test_login_link_is_permanent_and_reusable(client, edition, session, make_account, outbox): ana = make_volunteer(session, edition, first_name="Ana", email="ana@e.fr") make_account(Role.VOLUNTEER, volunteer_id=ana.id, email="ana@e.fr") client.post(_url(edition), json={"body": "{lien_connexion}"}) client.post(_url(edition), json={"body": "{lien_connexion}"}) first, second = (_link(m) for m in outbox) # Le même lien dans chaque email, utilisable plusieurs fois. assert first == second for _ in range(3): resp = _verify(client, edition, first) assert resp.status_code == 200 assert resp.json()["account"]["volunteer_id"] == str(ana.id) def test_reset_revokes_the_links_already_sent(client, edition, session, make_account, outbox): ana = make_volunteer(session, edition, first_name="Ana", email="ana@e.fr") account = make_account(Role.VOLUNTEER, volunteer_id=ana.id, email="ana@e.fr") client.post(_url(edition), json={"body": "{lien_connexion}"}) old = _link(outbox[0]) resp = client.post(f"/editions/{edition.id}/accounts/{account.id}/reset-login-link") assert resp.status_code == 200 assert _verify(client, edition, old).status_code == 401 client.post(_url(edition), json={"body": "{lien_connexion}"}) new = _link(outbox[1]) assert new != old assert _verify(client, edition, new).status_code == 200 def test_sending_the_link_gives_an_account_to_a_volunteer_without_one( client, edition, session, outbox ): from app.models import Account ana = make_volunteer(session, edition, first_name="Ana", email="ana@e.fr") client.post(_url(edition), json={"body": "{lien_connexion}"}) (account,) = session.exec(select(Account).where(Account.volunteer_id == ana.id)).all() assert account.role == Role.VOLUNTEER assert _verify(client, edition, _link(outbox[0])).status_code == 200 def test_shared_address_gets_the_login_page_only(client, edition, session, make_account, outbox): # Couple sur la même adresse : un seul compte possible pour cet email. ana = make_volunteer(session, edition, first_name="Ana", email="foyer@e.fr") make_account(Role.VOLUNTEER, volunteer_id=ana.id, email="foyer@e.fr") bob = make_volunteer(session, edition, first_name="Bob", email="foyer@e.fr") client.post(_url(edition), json={"volunteer_ids": [str(bob.id)], "body": "{lien_connexion}"}) assert "token=" not in plain_text(outbox[0]) def test_forged_or_coordinator_links_are_refused(client, edition, coordinator): from app.auth.permanent_link import permanent_login_token assert _verify(client, edition, permanent_login_token(coordinator)).status_code == 401 assert _verify(client, edition, "pl.nimportequoi.abc").status_code == 401 assert _verify(client, edition, f"pl.{coordinator.id.hex}.faux").status_code == 401