|
|
@@ -1,6 +1,7 @@
|
|
|
from datetime import datetime, timedelta
|
|
|
-from httpx import AsyncClient
|
|
|
+
|
|
|
import pytest
|
|
|
+from httpx import AsyncClient
|
|
|
from sqlalchemy import select
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
|
|
@@ -8,479 +9,549 @@ from app.main import app
|
|
|
from app.models import (
|
|
|
Commission,
|
|
|
CommissionMember,
|
|
|
+ OrgRole,
|
|
|
Project,
|
|
|
Slot,
|
|
|
SlotTemplate,
|
|
|
Sms,
|
|
|
User,
|
|
|
Volunteer,
|
|
|
- OrgRole,
|
|
|
)
|
|
|
from app.tests.conftest import default_sms_id, default_volunteer_id
|
|
|
|
|
|
+pytestmark = pytest.mark.asyncio
|
|
|
+
|
|
|
+WRITE_FORBIDDEN_ROLES = [OrgRole.RESPO_COMMISSION, None]
|
|
|
+
|
|
|
|
|
|
def get_arthur(session) -> Volunteer:
|
|
|
result = session.execute(select(Volunteer).where(Volunteer.id == default_volunteer_id))
|
|
|
arthur = result.scalars().first()
|
|
|
assert arthur is not None
|
|
|
return arthur
|
|
|
+
|
|
|
+
|
|
|
@pytest.fixture()
|
|
|
def user_headers(make_org_user) -> dict:
|
|
|
- _,headers = make_org_user(role=OrgRole.RESPO_BENEVOLE)
|
|
|
+ _, headers = make_org_user(role=OrgRole.RESPO_BENEVOLE)
|
|
|
return headers
|
|
|
-@pytest.fixture()
|
|
|
-
|
|
|
-async def test_create_sms_batch_bad_input(
|
|
|
- client: AsyncClient,
|
|
|
- user_headers: dict,
|
|
|
- session: Session,
|
|
|
- default_public_project: Project,
|
|
|
-):
|
|
|
- response = await client.post(
|
|
|
- app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
- json={"is_public": False},
|
|
|
- )
|
|
|
- assert response.status_code == 401
|
|
|
|
|
|
|
|
|
-async def test_basic(
|
|
|
- client: AsyncClient,
|
|
|
- user_headers: dict,
|
|
|
- session: Session,
|
|
|
- default_public_project: Project,
|
|
|
-):
|
|
|
- slot = Slot(
|
|
|
- project_id=default_public_project.id,
|
|
|
- title="replanter excalibure",
|
|
|
- starting_time=datetime.now() + timedelta(minutes=30),
|
|
|
- ending_time=datetime.now() + timedelta(minutes=60),
|
|
|
- )
|
|
|
- slot.volunteers.append(get_arthur(session))
|
|
|
- session.add(slot)
|
|
|
- session.commit()
|
|
|
- response = await client.post(
|
|
|
- app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
- headers=user_headers,
|
|
|
- json={"template": "Bonjour {prenom}!\n{titre}"},
|
|
|
- )
|
|
|
- assert response.status_code == 200
|
|
|
- result = session.execute(
|
|
|
- select(Sms).where(
|
|
|
- (Sms.project_id == default_public_project.id) & (Sms.id != default_sms_id)
|
|
|
+class TestCreateSmsBatchAuth:
|
|
|
+ async def test_requires_auth(self, client: AsyncClient, default_public_project: Project):
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
+ json={"is_public": False},
|
|
|
)
|
|
|
- )
|
|
|
- sms = result.scalars().first()
|
|
|
- assert sms is not None
|
|
|
- assert sms.content == "Bonjour Arthur!\nreplanter excalibure"
|
|
|
+ assert response.status_code == 401
|
|
|
+
|
|
|
+ async def test_forbidden_other_org(
|
|
|
+ self, client: AsyncClient, default_public_project: Project, other_org_user
|
|
|
+ ):
|
|
|
+ _, headers = other_org_user
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
+ json={"is_public": False},
|
|
|
+ headers=headers,
|
|
|
+ )
|
|
|
+ assert response.status_code == 403
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("role", [OrgRole.ORG_ADMIN, OrgRole.RESPO_BENEVOLE])
|
|
|
+ async def test_write_roles_allowed(
|
|
|
+ self, client: AsyncClient, default_public_project: Project, make_org_user, role
|
|
|
+ ):
|
|
|
+ _, headers = make_org_user(role=role)
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
+ headers=headers,
|
|
|
+ json={"template": "Bonjour!"},
|
|
|
+ )
|
|
|
+ assert response.status_code == 200
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("role", WRITE_FORBIDDEN_ROLES)
|
|
|
+ async def test_read_only_roles_forbidden(
|
|
|
+ self, client: AsyncClient, default_public_project: Project, make_org_user, role
|
|
|
+ ):
|
|
|
+ _, headers = make_org_user(role=role)
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
+ headers=headers,
|
|
|
+ json={"template": "Bonjour!"},
|
|
|
+ )
|
|
|
+ assert response.status_code == 403
|
|
|
+
|
|
|
+ async def test_other_org_member_forbidden(
|
|
|
+ self, client: AsyncClient, default_public_project: Project, other_org_user
|
|
|
+ ):
|
|
|
+ _, headers = other_org_user
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
+ headers=headers,
|
|
|
+ json={"template": "Bonjour!"},
|
|
|
+ )
|
|
|
+ assert response.status_code == 403
|
|
|
|
|
|
|
|
|
-async def test_multiple_volunteer(
|
|
|
- client: AsyncClient,
|
|
|
- user_headers: dict,
|
|
|
- session: Session,
|
|
|
- default_public_project: Project,
|
|
|
-):
|
|
|
- slot = Slot(
|
|
|
- project_id=default_public_project.id,
|
|
|
- title="replanter excalibure",
|
|
|
- starting_time=datetime.now() + timedelta(minutes=30),
|
|
|
- ending_time=datetime.now() + timedelta(minutes=60),
|
|
|
- )
|
|
|
- slot.volunteers.append(get_arthur(session))
|
|
|
- slot.volunteers.append(
|
|
|
- Volunteer(
|
|
|
+class TestCreateSmsBatchContent:
|
|
|
+ """Content-generation behavior -- exercised as respo_benevole throughout,
|
|
|
+ since role gating itself is covered separately in TestCreateSmsBatchAuth."""
|
|
|
+
|
|
|
+ async def test_basic(
|
|
|
+ self,
|
|
|
+ client: AsyncClient,
|
|
|
+ user_headers: dict,
|
|
|
+ session: Session,
|
|
|
+ default_public_project: Project,
|
|
|
+ ):
|
|
|
+ slot = Slot(
|
|
|
project_id=default_public_project.id,
|
|
|
- name="benevole 2",
|
|
|
- surname="echo",
|
|
|
- phone_number="t",
|
|
|
- email="",
|
|
|
- automatic_sms=True,
|
|
|
+ title="replanter excalibure",
|
|
|
+ starting_time=datetime.now() + timedelta(minutes=30),
|
|
|
+ ending_time=datetime.now() + timedelta(minutes=60),
|
|
|
)
|
|
|
- )
|
|
|
- session.add(slot)
|
|
|
- session.commit()
|
|
|
- response = await client.post(
|
|
|
- app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
- headers=user_headers,
|
|
|
- json={"template": "Bonjour {prenom}!\n{titre}"},
|
|
|
- )
|
|
|
- assert response.status_code == 200
|
|
|
- result = session.execute(
|
|
|
- select(Sms).where(
|
|
|
- (Sms.project_id == default_public_project.id) & (Sms.id != default_sms_id)
|
|
|
+ slot.volunteers.append(get_arthur(session))
|
|
|
+ session.add(slot)
|
|
|
+ session.commit()
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
+ headers=user_headers,
|
|
|
+ json={"template": "Bonjour {prenom}!\n{titre}"},
|
|
|
)
|
|
|
- )
|
|
|
- count_sms = sum(1 for _ in result.fetchall())
|
|
|
- assert count_sms == 2
|
|
|
-
|
|
|
-
|
|
|
-@pytest.mark.parametrize("automatic_sms,expected", [(True, True), (False, False)])
|
|
|
-async def test_volunteer_settings(
|
|
|
- client: AsyncClient,
|
|
|
- user_headers: dict,
|
|
|
- session: Session,
|
|
|
- default_public_project: Project,
|
|
|
- automatic_sms: bool,
|
|
|
- expected: bool,
|
|
|
-):
|
|
|
- slot = Slot(
|
|
|
- project_id=default_public_project.id,
|
|
|
- title="replanter excalibure",
|
|
|
- starting_time=datetime.now() + timedelta(minutes=30),
|
|
|
- ending_time=datetime.now() + timedelta(minutes=60),
|
|
|
- )
|
|
|
- arthur = get_arthur(session)
|
|
|
- arthur.automatic_sms = automatic_sms
|
|
|
- slot.volunteers.append(arthur)
|
|
|
- session.add(slot)
|
|
|
- session.commit()
|
|
|
- response = await client.post(
|
|
|
- app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
- headers=user_headers,
|
|
|
- json={"template": "Bonjour {prenom}!\n{titre}"},
|
|
|
- )
|
|
|
- assert response.status_code == 200
|
|
|
- result = session.execute(
|
|
|
- select(Sms).where(
|
|
|
- (Sms.project_id == default_public_project.id) & (Sms.id != default_sms_id)
|
|
|
+ assert response.status_code == 200
|
|
|
+ sms = (
|
|
|
+ session.execute(
|
|
|
+ select(Sms).where(
|
|
|
+ (Sms.project_id == default_public_project.id) & (Sms.id != default_sms_id)
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .scalars()
|
|
|
+ .first()
|
|
|
)
|
|
|
- )
|
|
|
- sms = result.scalars().first()
|
|
|
- assert (sms is not None) == expected
|
|
|
-
|
|
|
-
|
|
|
-async def test_only_future_slot(
|
|
|
- client: AsyncClient,
|
|
|
- user_headers: dict,
|
|
|
- session: Session,
|
|
|
- default_public_project: Project,
|
|
|
-):
|
|
|
- slot = Slot(
|
|
|
- project_id=default_public_project.id,
|
|
|
- title="replanter excalibure",
|
|
|
- starting_time=datetime.now() + timedelta(minutes=-30),
|
|
|
- ending_time=datetime.now() + timedelta(minutes=60),
|
|
|
- )
|
|
|
- slot.volunteers.append(get_arthur(session))
|
|
|
- session.add(slot)
|
|
|
- session.commit()
|
|
|
- response = await client.post(
|
|
|
- app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
- headers=user_headers,
|
|
|
- json={"template": "Bonjour {prenom}!\n{titre}"},
|
|
|
- )
|
|
|
- assert response.status_code == 200
|
|
|
- result = session.execute(
|
|
|
- select(Sms).where(
|
|
|
- (Sms.project_id == default_public_project.id) & (Sms.id != default_sms_id)
|
|
|
+ assert sms is not None
|
|
|
+ assert sms.content == "Bonjour Arthur!\nreplanter excalibure"
|
|
|
+
|
|
|
+ async def test_multiple_volunteer(
|
|
|
+ self,
|
|
|
+ client: AsyncClient,
|
|
|
+ user_headers: dict,
|
|
|
+ session: Session,
|
|
|
+ default_public_project: Project,
|
|
|
+ ):
|
|
|
+ slot = Slot(
|
|
|
+ project_id=default_public_project.id,
|
|
|
+ title="replanter excalibure",
|
|
|
+ starting_time=datetime.now() + timedelta(minutes=30),
|
|
|
+ ending_time=datetime.now() + timedelta(minutes=60),
|
|
|
)
|
|
|
- )
|
|
|
- sms = result.scalars().first()
|
|
|
- assert sms is None
|
|
|
-
|
|
|
-
|
|
|
-@pytest.mark.parametrize("delta_t,expected", [(32, False), (28, True)])
|
|
|
-async def test_time_delta(
|
|
|
- client: AsyncClient,
|
|
|
- user_headers: dict,
|
|
|
- session: Session,
|
|
|
- default_public_project: Project,
|
|
|
- delta_t: int,
|
|
|
- expected,
|
|
|
-):
|
|
|
- slot = Slot(
|
|
|
- project_id=default_public_project.id,
|
|
|
- title="replanter excalibure",
|
|
|
- starting_time=datetime.now() + timedelta(minutes=30),
|
|
|
- ending_time=datetime.now() + timedelta(minutes=60),
|
|
|
- )
|
|
|
- slot.volunteers.append(get_arthur(session))
|
|
|
- session.add(slot)
|
|
|
- session.commit()
|
|
|
- response = await client.post(
|
|
|
- app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
- headers=user_headers,
|
|
|
- json={"template": "Bonjour {prenom}!\n{titre}", "delta_t": delta_t},
|
|
|
- )
|
|
|
- assert response.status_code == 200
|
|
|
- result = session.execute(
|
|
|
- select(Sms).where(
|
|
|
- (Sms.project_id == default_public_project.id) & (Sms.id != default_sms_id)
|
|
|
+ slot.volunteers.append(get_arthur(session))
|
|
|
+ slot.volunteers.append(
|
|
|
+ Volunteer(
|
|
|
+ project_id=default_public_project.id,
|
|
|
+ name="benevole 2",
|
|
|
+ surname="echo",
|
|
|
+ phone_number="t",
|
|
|
+ email="",
|
|
|
+ automatic_sms=True,
|
|
|
+ )
|
|
|
)
|
|
|
- )
|
|
|
- sms = result.scalars().first()
|
|
|
- assert (sms is not None) == expected
|
|
|
+ session.add(slot)
|
|
|
+ session.commit()
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
+ headers=user_headers,
|
|
|
+ json={"template": "Bonjour {prenom}!\n{titre}"},
|
|
|
+ )
|
|
|
+ assert response.status_code == 200
|
|
|
+ result = session.execute(
|
|
|
+ select(Sms).where(
|
|
|
+ (Sms.project_id == default_public_project.id) & (Sms.id != default_sms_id)
|
|
|
+ )
|
|
|
+ )
|
|
|
+ count_sms = sum(1 for _ in result.fetchall())
|
|
|
+ assert count_sms == 2
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("automatic_sms,expected", [(True, True), (False, False)])
|
|
|
+ async def test_volunteer_settings(
|
|
|
+ self,
|
|
|
+ client: AsyncClient,
|
|
|
+ user_headers: dict,
|
|
|
+ session: Session,
|
|
|
+ default_public_project: Project,
|
|
|
+ automatic_sms: bool,
|
|
|
+ expected: bool,
|
|
|
+ ):
|
|
|
+ slot = Slot(
|
|
|
+ project_id=default_public_project.id,
|
|
|
+ title="replanter excalibure",
|
|
|
+ starting_time=datetime.now() + timedelta(minutes=30),
|
|
|
+ ending_time=datetime.now() + timedelta(minutes=60),
|
|
|
+ )
|
|
|
+ arthur = get_arthur(session)
|
|
|
+ arthur.automatic_sms = automatic_sms
|
|
|
+ slot.volunteers.append(arthur)
|
|
|
+ session.add(slot)
|
|
|
+ session.commit()
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
+ headers=user_headers,
|
|
|
+ json={"template": "Bonjour {prenom}!\n{titre}"},
|
|
|
+ )
|
|
|
+ assert response.status_code == 200
|
|
|
+ sms = (
|
|
|
+ session.execute(
|
|
|
+ select(Sms).where(
|
|
|
+ (Sms.project_id == default_public_project.id) & (Sms.id != default_sms_id)
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .scalars()
|
|
|
+ .first()
|
|
|
+ )
|
|
|
+ assert (sms is not None) == expected
|
|
|
+
|
|
|
+ async def test_only_future_slot(
|
|
|
+ self,
|
|
|
+ client: AsyncClient,
|
|
|
+ user_headers: dict,
|
|
|
+ session: Session,
|
|
|
+ default_public_project: Project,
|
|
|
+ ):
|
|
|
+ slot = Slot(
|
|
|
+ project_id=default_public_project.id,
|
|
|
+ title="replanter excalibure",
|
|
|
+ starting_time=datetime.now() + timedelta(minutes=-30),
|
|
|
+ ending_time=datetime.now() + timedelta(minutes=60),
|
|
|
+ )
|
|
|
+ slot.volunteers.append(get_arthur(session))
|
|
|
+ session.add(slot)
|
|
|
+ session.commit()
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
+ headers=user_headers,
|
|
|
+ json={"template": "Bonjour {prenom}!\n{titre}"},
|
|
|
+ )
|
|
|
+ assert response.status_code == 200
|
|
|
+ sms = (
|
|
|
+ session.execute(
|
|
|
+ select(Sms).where(
|
|
|
+ (Sms.project_id == default_public_project.id) & (Sms.id != default_sms_id)
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .scalars()
|
|
|
+ .first()
|
|
|
+ )
|
|
|
+ assert sms is None
|
|
|
+
|
|
|
+ @pytest.mark.parametrize("delta_t,expected", [(32, False), (28, True)])
|
|
|
+ async def test_time_delta(
|
|
|
+ self,
|
|
|
+ client: AsyncClient,
|
|
|
+ user_headers: dict,
|
|
|
+ session: Session,
|
|
|
+ default_public_project: Project,
|
|
|
+ delta_t: int,
|
|
|
+ expected,
|
|
|
+ ):
|
|
|
+ slot = Slot(
|
|
|
+ project_id=default_public_project.id,
|
|
|
+ title="replanter excalibure",
|
|
|
+ starting_time=datetime.now() + timedelta(minutes=30),
|
|
|
+ ending_time=datetime.now() + timedelta(minutes=60),
|
|
|
+ )
|
|
|
+ slot.volunteers.append(get_arthur(session))
|
|
|
+ session.add(slot)
|
|
|
+ session.commit()
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
+ headers=user_headers,
|
|
|
+ json={"template": "Bonjour {prenom}!\n{titre}", "delta_t": delta_t},
|
|
|
+ )
|
|
|
+ assert response.status_code == 200
|
|
|
+ sms = (
|
|
|
+ session.execute(
|
|
|
+ select(Sms).where(
|
|
|
+ (Sms.project_id == default_public_project.id) & (Sms.id != default_sms_id)
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .scalars()
|
|
|
+ .first()
|
|
|
+ )
|
|
|
+ assert (sms is not None) == expected
|
|
|
|
|
|
|
|
|
starting_time = datetime.now() + timedelta(minutes=30)
|
|
|
|
|
|
|
|
|
-@pytest.mark.parametrize(
|
|
|
- "template,expected",
|
|
|
- [
|
|
|
- ("no escape", "no escape"),
|
|
|
- ("{titre}", "replanter excalibure"),
|
|
|
- ("{prenom}", "Arthur"),
|
|
|
- ("{prenom} {nom}", "Arthur Pandragon"),
|
|
|
- ("{debut}", starting_time.strftime("%Hh%M")),
|
|
|
- (
|
|
|
- "{titre} {description}, {respo}",
|
|
|
- "replanter excalibure {description}, {respo}",
|
|
|
- ),
|
|
|
- ],
|
|
|
-)
|
|
|
-async def test_content_no_template(
|
|
|
- client: AsyncClient,
|
|
|
- user_headers: dict,
|
|
|
- session: Session,
|
|
|
- default_public_project: Project,
|
|
|
- template: str,
|
|
|
- expected: str,
|
|
|
-):
|
|
|
- """No SlotTemplate attached at all -- {respo}/{description} placeholders
|
|
|
- are left untouched, since the replace only happens when slot.template
|
|
|
- is not None. Unaffected by the commission/override change."""
|
|
|
- slot = Slot(
|
|
|
- project_id=default_public_project.id,
|
|
|
- title="replanter excalibure",
|
|
|
- starting_time=starting_time,
|
|
|
- ending_time=datetime.now() + timedelta(minutes=60),
|
|
|
- )
|
|
|
- slot.volunteers.append(get_arthur(session))
|
|
|
- session.add(slot)
|
|
|
- session.commit()
|
|
|
- response = await client.post(
|
|
|
- app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
- headers=user_headers,
|
|
|
- json={"template": template},
|
|
|
- )
|
|
|
- assert response.status_code == 200
|
|
|
- result = session.execute(
|
|
|
- select(Sms).where(
|
|
|
- (Sms.project_id == default_public_project.id) & (Sms.id != default_sms_id)
|
|
|
+class TestSmsPlaceholders:
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "template,expected",
|
|
|
+ [
|
|
|
+ ("no escape", "no escape"),
|
|
|
+ ("{titre}", "replanter excalibure"),
|
|
|
+ ("{prenom}", "Arthur"),
|
|
|
+ ("{prenom} {nom}", "Arthur Pandragon"),
|
|
|
+ ("{debut}", starting_time.strftime("%Hh%M")),
|
|
|
+ ("{titre} {description}, {respo}", "replanter excalibure {description}, {respo}"),
|
|
|
+ ],
|
|
|
+ )
|
|
|
+ async def test_content_no_template(
|
|
|
+ self,
|
|
|
+ client: AsyncClient,
|
|
|
+ user_headers: dict,
|
|
|
+ session: Session,
|
|
|
+ default_public_project: Project,
|
|
|
+ template: str,
|
|
|
+ expected: str,
|
|
|
+ ):
|
|
|
+ """No SlotTemplate attached at all -- {respo}/{description} placeholders
|
|
|
+ are left untouched, since the replace only happens when slot.template
|
|
|
+ is not None."""
|
|
|
+ slot = Slot(
|
|
|
+ project_id=default_public_project.id,
|
|
|
+ title="replanter excalibure",
|
|
|
+ starting_time=starting_time,
|
|
|
+ ending_time=datetime.now() + timedelta(minutes=60),
|
|
|
)
|
|
|
- )
|
|
|
- sms = result.scalars().first()
|
|
|
- assert sms is not None
|
|
|
- assert sms.content == expected
|
|
|
-
|
|
|
-
|
|
|
-@pytest.mark.parametrize(
|
|
|
- "template,expected",
|
|
|
- [
|
|
|
- ("no escape", "no escape"),
|
|
|
- ("{titre}", "replanter excalibure"),
|
|
|
- ("{prenom}", "Arthur"),
|
|
|
- ("{prenom} {nom}", "Arthur Pandragon"),
|
|
|
- ("{debut}", starting_time.strftime("%Hh%M")),
|
|
|
- ],
|
|
|
-)
|
|
|
-async def test_content_with_template_non_respo_placeholders(
|
|
|
- client: AsyncClient,
|
|
|
- user_headers: dict,
|
|
|
- session: Session,
|
|
|
- default_public_project: Project,
|
|
|
- template: str,
|
|
|
- expected: str,
|
|
|
-):
|
|
|
- """Placeholders unrelated to {respo} behave the same regardless of how
|
|
|
- the contact is resolved -- kept separate from the {respo}-specific cases
|
|
|
- below since those need commission/override setup."""
|
|
|
- slot_template = SlotTemplate(
|
|
|
- project_id=default_public_project.id,
|
|
|
- title="replanter excalibure",
|
|
|
- description="rendre l'épee à son caillou",
|
|
|
- place="gros caillou",
|
|
|
- )
|
|
|
- slot = Slot(
|
|
|
- project_id=default_public_project.id,
|
|
|
- title="replanter excalibure",
|
|
|
- starting_time=starting_time,
|
|
|
- ending_time=datetime.now() + timedelta(minutes=60),
|
|
|
- )
|
|
|
- slot.template = slot_template
|
|
|
- slot.volunteers.append(get_arthur(session))
|
|
|
- session.add(slot_template)
|
|
|
- session.add(slot)
|
|
|
- session.commit()
|
|
|
-
|
|
|
- response = await client.post(
|
|
|
- app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
- headers=user_headers,
|
|
|
- json={"template": template},
|
|
|
- )
|
|
|
- assert response.status_code == 200
|
|
|
- result = session.execute(
|
|
|
- select(Sms).where(
|
|
|
- (Sms.project_id == default_public_project.id) & (Sms.id != default_sms_id)
|
|
|
+ slot.volunteers.append(get_arthur(session))
|
|
|
+ session.add(slot)
|
|
|
+ session.commit()
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
+ headers=user_headers,
|
|
|
+ json={"template": template},
|
|
|
)
|
|
|
- )
|
|
|
- sms = result.scalars().first()
|
|
|
- assert sms is not None
|
|
|
- assert sms.content == expected
|
|
|
-
|
|
|
-
|
|
|
-async def test_respo_placeholder_uses_override_when_set(
|
|
|
- client: AsyncClient,
|
|
|
- user_headers: dict,
|
|
|
- session: Session,
|
|
|
- default_public_project: Project,
|
|
|
-):
|
|
|
- """responsible_override wins even if a commission is also attached."""
|
|
|
- commission = Commission(project_id=default_public_project.id, name="Bar")
|
|
|
- session.add(commission)
|
|
|
- session.commit()
|
|
|
-
|
|
|
- slot_template = SlotTemplate(
|
|
|
- project_id=default_public_project.id,
|
|
|
- title="replanter excalibure",
|
|
|
- description="rendre l'épee à son caillou",
|
|
|
- place="gros caillou",
|
|
|
- commission_id=commission.id,
|
|
|
- responsible_override="dieu",
|
|
|
- )
|
|
|
- slot = Slot(
|
|
|
- project_id=default_public_project.id,
|
|
|
- title="replanter excalibure",
|
|
|
- starting_time=starting_time,
|
|
|
- ending_time=datetime.now() + timedelta(minutes=60),
|
|
|
- )
|
|
|
- slot.template = slot_template
|
|
|
- slot.volunteers.append(get_arthur(session))
|
|
|
- session.add(slot_template)
|
|
|
- session.add(slot)
|
|
|
- session.commit()
|
|
|
-
|
|
|
- response = await client.post(
|
|
|
- app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
- headers=user_headers,
|
|
|
- json={"template": "{titre} {description}, {respo}"},
|
|
|
- )
|
|
|
- assert response.status_code == 200
|
|
|
- result = session.execute(
|
|
|
- select(Sms).where(
|
|
|
- (Sms.project_id == default_public_project.id) & (Sms.id != default_sms_id)
|
|
|
+ assert response.status_code == 200
|
|
|
+ sms = (
|
|
|
+ session.execute(
|
|
|
+ select(Sms).where(
|
|
|
+ (Sms.project_id == default_public_project.id) & (Sms.id != default_sms_id)
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .scalars()
|
|
|
+ .first()
|
|
|
)
|
|
|
- )
|
|
|
- sms = result.scalars().first()
|
|
|
- assert sms is not None
|
|
|
- assert sms.content == "replanter excalibure rendre l'épee à son caillou, dieu"
|
|
|
-
|
|
|
+ assert sms is not None
|
|
|
+ assert sms.content == expected
|
|
|
|
|
|
-async def test_respo_placeholder_derives_from_commission_members(
|
|
|
- client: AsyncClient,
|
|
|
- user_headers: dict,
|
|
|
- session: Session,
|
|
|
- default_public_project: Project,
|
|
|
-):
|
|
|
- """No override set -- {respo} falls back to the commission's members'
|
|
|
- name : phone_number, comma-joined."""
|
|
|
- commission = Commission(project_id=default_public_project.id, name="Bar")
|
|
|
- session.add(commission)
|
|
|
- session.commit()
|
|
|
-
|
|
|
- member1 = User(
|
|
|
- id="c1a1a1a1-0000-0000-0000-000000000001",
|
|
|
- email="merlin@test.com",
|
|
|
- hashed_password="x",
|
|
|
- name="Merlin",
|
|
|
- phone_number="0601020304",
|
|
|
- )
|
|
|
- member2 = User(
|
|
|
- id="c1a1a1a1-0000-0000-0000-000000000002",
|
|
|
- email="morgane@test.com",
|
|
|
- hashed_password="x",
|
|
|
- name="Morgane",
|
|
|
- phone_number="0605060708",
|
|
|
- )
|
|
|
- session.add_all([member1, member2])
|
|
|
- session.commit()
|
|
|
- session.add_all(
|
|
|
+ @pytest.mark.parametrize(
|
|
|
+ "template,expected",
|
|
|
[
|
|
|
- CommissionMember(commission_id=commission.id, user_id=member1.id),
|
|
|
- CommissionMember(commission_id=commission.id, user_id=member2.id),
|
|
|
- ]
|
|
|
- )
|
|
|
- session.commit()
|
|
|
-
|
|
|
- slot_template = SlotTemplate(
|
|
|
- project_id=default_public_project.id,
|
|
|
- title="replanter excalibure",
|
|
|
- description="rendre l'épee à son caillou",
|
|
|
- place="gros caillou",
|
|
|
- commission_id=commission.id,
|
|
|
- # responsible_override intentionally left unset
|
|
|
- )
|
|
|
- slot = Slot(
|
|
|
- project_id=default_public_project.id,
|
|
|
- title="replanter excalibure",
|
|
|
- starting_time=starting_time,
|
|
|
- ending_time=datetime.now() + timedelta(minutes=60),
|
|
|
- )
|
|
|
- slot.template = slot_template
|
|
|
- slot.volunteers.append(get_arthur(session))
|
|
|
- session.add(slot_template)
|
|
|
- session.add(slot)
|
|
|
- session.commit()
|
|
|
-
|
|
|
- response = await client.post(
|
|
|
- app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
- headers=user_headers,
|
|
|
- json={"template": "{titre} {description}, {respo}"},
|
|
|
- )
|
|
|
- assert response.status_code == 200
|
|
|
- result = session.execute(
|
|
|
- select(Sms).where(
|
|
|
- (Sms.project_id == default_public_project.id) & (Sms.id != default_sms_id)
|
|
|
+ ("no escape", "no escape"),
|
|
|
+ ("{titre}", "replanter excalibure"),
|
|
|
+ ("{prenom}", "Arthur"),
|
|
|
+ ("{prenom} {nom}", "Arthur Pandragon"),
|
|
|
+ ("{debut}", starting_time.strftime("%Hh%M")),
|
|
|
+ ],
|
|
|
+ )
|
|
|
+ async def test_content_with_template_non_respo_placeholders(
|
|
|
+ self,
|
|
|
+ client: AsyncClient,
|
|
|
+ user_headers: dict,
|
|
|
+ session: Session,
|
|
|
+ default_public_project: Project,
|
|
|
+ template: str,
|
|
|
+ expected: str,
|
|
|
+ ):
|
|
|
+ """Placeholders unrelated to {respo} behave the same regardless of
|
|
|
+ how the contact is resolved."""
|
|
|
+ slot_template = SlotTemplate(
|
|
|
+ project_id=default_public_project.id,
|
|
|
+ title="replanter excalibure",
|
|
|
+ description="rendre l'épee à son caillou",
|
|
|
+ place="gros caillou",
|
|
|
+ )
|
|
|
+ slot = Slot(
|
|
|
+ project_id=default_public_project.id,
|
|
|
+ title="replanter excalibure",
|
|
|
+ starting_time=starting_time,
|
|
|
+ ending_time=datetime.now() + timedelta(minutes=60),
|
|
|
+ )
|
|
|
+ slot.template = slot_template
|
|
|
+ slot.volunteers.append(get_arthur(session))
|
|
|
+ session.add(slot_template)
|
|
|
+ session.add(slot)
|
|
|
+ session.commit()
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
+ headers=user_headers,
|
|
|
+ json={"template": template},
|
|
|
+ )
|
|
|
+ assert response.status_code == 200
|
|
|
+ sms = (
|
|
|
+ session.execute(
|
|
|
+ select(Sms).where(
|
|
|
+ (Sms.project_id == default_public_project.id) & (Sms.id != default_sms_id)
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .scalars()
|
|
|
+ .first()
|
|
|
+ )
|
|
|
+ assert sms is not None
|
|
|
+ assert sms.content == expected
|
|
|
+
|
|
|
+ async def test_respo_placeholder_uses_override_when_set(
|
|
|
+ self,
|
|
|
+ client: AsyncClient,
|
|
|
+ user_headers: dict,
|
|
|
+ session: Session,
|
|
|
+ default_public_project: Project,
|
|
|
+ ):
|
|
|
+ """responsible_override wins even if a commission is also attached."""
|
|
|
+ commission = Commission(project_id=default_public_project.id, name="Bar")
|
|
|
+ session.add(commission)
|
|
|
+ session.commit()
|
|
|
+ slot_template = SlotTemplate(
|
|
|
+ project_id=default_public_project.id,
|
|
|
+ title="replanter excalibure",
|
|
|
+ description="rendre l'épee à son caillou",
|
|
|
+ place="gros caillou",
|
|
|
+ commission_id=commission.id,
|
|
|
+ responsible_override="dieu",
|
|
|
+ )
|
|
|
+ slot = Slot(
|
|
|
+ project_id=default_public_project.id,
|
|
|
+ title="replanter excalibure",
|
|
|
+ starting_time=starting_time,
|
|
|
+ ending_time=datetime.now() + timedelta(minutes=60),
|
|
|
+ )
|
|
|
+ slot.template = slot_template
|
|
|
+ slot.volunteers.append(get_arthur(session))
|
|
|
+ session.add(slot_template)
|
|
|
+ session.add(slot)
|
|
|
+ session.commit()
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
+ headers=user_headers,
|
|
|
+ json={"template": "{titre} {description}, {respo}"},
|
|
|
+ )
|
|
|
+ assert response.status_code == 200
|
|
|
+ sms = (
|
|
|
+ session.execute(
|
|
|
+ select(Sms).where(
|
|
|
+ (Sms.project_id == default_public_project.id) & (Sms.id != default_sms_id)
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .scalars()
|
|
|
+ .first()
|
|
|
+ )
|
|
|
+ assert sms is not None
|
|
|
+ assert sms.content == "replanter excalibure rendre l'épee à son caillou, dieu"
|
|
|
+
|
|
|
+ async def test_respo_placeholder_derives_from_commission_members(
|
|
|
+ self,
|
|
|
+ client: AsyncClient,
|
|
|
+ user_headers: dict,
|
|
|
+ session: Session,
|
|
|
+ default_public_project: Project,
|
|
|
+ ):
|
|
|
+ """No override set -- {respo} falls back to the commission's members'
|
|
|
+ name : phone_number, comma-joined."""
|
|
|
+ commission = Commission(project_id=default_public_project.id, name="Bar")
|
|
|
+ session.add(commission)
|
|
|
+ session.commit()
|
|
|
+ member1 = User(
|
|
|
+ id="c1a1a1a1-0000-0000-0000-000000000001",
|
|
|
+ email="merlin@test.com",
|
|
|
+ hashed_password="x",
|
|
|
+ name="Merlin",
|
|
|
+ phone_number="0601020304",
|
|
|
+ )
|
|
|
+ member2 = User(
|
|
|
+ id="c1a1a1a1-0000-0000-0000-000000000002",
|
|
|
+ email="morgane@test.com",
|
|
|
+ hashed_password="x",
|
|
|
+ name="Morgane",
|
|
|
+ phone_number="0605060708",
|
|
|
+ )
|
|
|
+ session.add_all([member1, member2])
|
|
|
+ session.commit()
|
|
|
+ session.add_all(
|
|
|
+ [
|
|
|
+ CommissionMember(commission_id=commission.id, user_id=member1.id),
|
|
|
+ CommissionMember(commission_id=commission.id, user_id=member2.id),
|
|
|
+ ]
|
|
|
+ )
|
|
|
+ session.commit()
|
|
|
+ slot_template = SlotTemplate(
|
|
|
+ project_id=default_public_project.id,
|
|
|
+ title="replanter excalibure",
|
|
|
+ description="rendre l'épee à son caillou",
|
|
|
+ place="gros caillou",
|
|
|
+ commission_id=commission.id,
|
|
|
+ )
|
|
|
+ slot = Slot(
|
|
|
+ project_id=default_public_project.id,
|
|
|
+ title="replanter excalibure",
|
|
|
+ starting_time=starting_time,
|
|
|
+ ending_time=datetime.now() + timedelta(minutes=60),
|
|
|
+ )
|
|
|
+ slot.template = slot_template
|
|
|
+ slot.volunteers.append(get_arthur(session))
|
|
|
+ session.add(slot_template)
|
|
|
+ session.add(slot)
|
|
|
+ session.commit()
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
+ headers=user_headers,
|
|
|
+ json={"template": "{titre} {description}, {respo}"},
|
|
|
+ )
|
|
|
+ assert response.status_code == 200
|
|
|
+ sms = (
|
|
|
+ session.execute(
|
|
|
+ select(Sms).where(
|
|
|
+ (Sms.project_id == default_public_project.id) & (Sms.id != default_sms_id)
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .scalars()
|
|
|
+ .first()
|
|
|
+ )
|
|
|
+ assert sms is not None
|
|
|
+ assert sms.content == (
|
|
|
+ "replanter excalibure rendre l'épee à son caillou, "
|
|
|
+ "Merlin : 0601020304, Morgane : 0605060708"
|
|
|
)
|
|
|
- )
|
|
|
- sms = result.scalars().first()
|
|
|
- assert sms is not None
|
|
|
- assert sms.content == (
|
|
|
- "replanter excalibure rendre l'épee à son caillou, "
|
|
|
- "Merlin : 0601020304, Morgane : 0605060708"
|
|
|
- )
|
|
|
-
|
|
|
-
|
|
|
-async def test_respo_placeholder_empty_when_no_commission_or_override(
|
|
|
- client: AsyncClient,
|
|
|
- user_headers: dict,
|
|
|
- session: Session,
|
|
|
- default_public_project: Project,
|
|
|
-):
|
|
|
- """Template attached but no commission and no override -- {respo}
|
|
|
- resolves to an empty string rather than leaving the placeholder literal
|
|
|
- (that untouched-placeholder behavior is only for slot.template is None,
|
|
|
- covered in test_content_no_template)."""
|
|
|
- slot_template = SlotTemplate(
|
|
|
- project_id=default_public_project.id,
|
|
|
- title="replanter excalibure",
|
|
|
- description="rendre l'épee à son caillou",
|
|
|
- place="gros caillou",
|
|
|
- )
|
|
|
- slot = Slot(
|
|
|
- project_id=default_public_project.id,
|
|
|
- title="replanter excalibure",
|
|
|
- starting_time=starting_time,
|
|
|
- ending_time=datetime.now() + timedelta(minutes=60),
|
|
|
- )
|
|
|
- slot.template = slot_template
|
|
|
- slot.volunteers.append(get_arthur(session))
|
|
|
- session.add(slot_template)
|
|
|
- session.add(slot)
|
|
|
- session.commit()
|
|
|
|
|
|
- response = await client.post(
|
|
|
- app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
- headers=user_headers,
|
|
|
- json={"template": "{titre}, {respo}"},
|
|
|
- )
|
|
|
- assert response.status_code == 200
|
|
|
- result = session.execute(
|
|
|
- select(Sms).where(
|
|
|
- (Sms.project_id == default_public_project.id) & (Sms.id != default_sms_id)
|
|
|
+ async def test_respo_placeholder_empty_when_no_commission_or_override(
|
|
|
+ self,
|
|
|
+ client: AsyncClient,
|
|
|
+ user_headers: dict,
|
|
|
+ session: Session,
|
|
|
+ default_public_project: Project,
|
|
|
+ ):
|
|
|
+ """Template attached but no commission and no override -- {respo}
|
|
|
+ resolves to an empty string rather than leaving the placeholder
|
|
|
+ literal (that untouched-placeholder behavior only applies when
|
|
|
+ slot.template is None, covered in test_content_no_template)."""
|
|
|
+ slot_template = SlotTemplate(
|
|
|
+ project_id=default_public_project.id,
|
|
|
+ title="replanter excalibure",
|
|
|
+ description="rendre l'épee à son caillou",
|
|
|
+ place="gros caillou",
|
|
|
)
|
|
|
- )
|
|
|
- sms = result.scalars().first()
|
|
|
- assert sms is not None
|
|
|
- assert sms.content == "replanter excalibure, "
|
|
|
+ slot = Slot(
|
|
|
+ project_id=default_public_project.id,
|
|
|
+ title="replanter excalibure",
|
|
|
+ starting_time=starting_time,
|
|
|
+ ending_time=datetime.now() + timedelta(minutes=60),
|
|
|
+ )
|
|
|
+ slot.template = slot_template
|
|
|
+ slot.volunteers.append(get_arthur(session))
|
|
|
+ session.add(slot_template)
|
|
|
+ session.add(slot)
|
|
|
+ session.commit()
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_sms_batch", project_id=default_public_project.id),
|
|
|
+ headers=user_headers,
|
|
|
+ json={"template": "{titre}, {respo}"},
|
|
|
+ )
|
|
|
+ assert response.status_code == 200
|
|
|
+ sms = (
|
|
|
+ session.execute(
|
|
|
+ select(Sms).where(
|
|
|
+ (Sms.project_id == default_public_project.id) & (Sms.id != default_sms_id)
|
|
|
+ )
|
|
|
+ )
|
|
|
+ .scalars()
|
|
|
+ .first()
|
|
|
+ )
|
|
|
+ assert sms is not None
|
|
|
+ assert sms.content == "replanter excalibure, "
|