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

from datetime import datetime, timedelta
from httpx import AsyncClient
import pytest
from sqlalchemy import select
from sqlalchemy.orm import Session

from app.main import app
from app.models import (
    Commission,
    CommissionMember,
    Project,
    Slot,
    SlotTemplate,
    Sms,
    User,
    Volunteer,
    OrgRole,
)
from app.tests.conftest import default_sms_id, default_volunteer_id


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
@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)
        )
    )
    sms = result.scalars().first()
    assert sms is not None
    assert sms.content == "Bonjour Arthur!\nreplanter excalibure"


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(
            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(
    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)
        )
    )
    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)
        )
    )
    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)
        )
    )
    sms = result.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)
        )
    )
    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)
        )
    )
    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)
        )
    )
    sms = result.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(
    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,
        # 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)
        )
    )
    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)
        )
    )
    sms = result.scalars().first()
    assert sms is not None
    assert sms.content == "replanter excalibure, "