bdlg2023-sms-back/app/tests/test_sms_batch.py

from datetime import datetime, timedelta

import pytest
from httpx import AsyncClient
from sqlalchemy import select
from sqlalchemy.orm import Session

from app.main import app
from app.models import (
    Commission,
    CommissionMember,
    OrgRole,
    Project,
    Slot,
    SlotTemplate,
    Sms,
    User,
    Volunteer,
)
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)
    return headers


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},
        )
        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


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,
            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 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),
        )
        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,
            )
        )
        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)


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),
        )
        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
        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

    @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(
        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"
        )

    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",
        )
        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, "