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 Project, Slot, SlotTemplate, Sms, Volunteer
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


async def test_create_sms_batch_bad_input(
    client: AsyncClient,
    default_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,
    default_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()
    # créer un sms simple
    response = await client.post(
        app.url_path_for("create_sms_batch", project_id=default_public_project.id),
        headers=default_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,
    default_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()
    # créer un sms simple
    response = await client.post(
        app.url_path_for("create_sms_batch", project_id=default_public_project.id),
        headers=default_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 = 0
    for _ in result.fetchall():
        count_sms += 1
    assert count_sms == 2


@pytest.mark.parametrize("automatic_sms,expected", [(True, True), (False, False)])
async def test_volunteer_settings(
    client: AsyncClient,
    default_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()
    # créer un sms simple
    response = await client.post(
        app.url_path_for("create_sms_batch", project_id=default_public_project.id),
        headers=default_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,
    default_user_headers: dict,
    session: Session,
    default_public_project: Project,
):
    # Create a slot starting in the past
    # no sms should be issued
    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()

    # créer un sms simple
    response = await client.post(
        app.url_path_for("create_sms_batch", project_id=default_public_project.id),
        headers=default_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,
    default_user_headers: dict,
    session: Session,
    default_public_project: Project,
    delta_t: int,
    expected,
):
    # Create a slot starting in the past
    # no sms should be issued
    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()

    # créer un sms simple
    response = await client.post(
        app.url_path_for("create_sms_batch", project_id=default_public_project.id),
        headers=default_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,
    default_user_headers: dict,
    session: Session,
    default_public_project: Project,
    template: str,
    expected: str,
):
    # Create a slot starting in the past
    # no sms should be issued
    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()

    # créer un sms simple
    response = await client.post(
        app.url_path_for("create_sms_batch", project_id=default_public_project.id),
        headers=default_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")),
        (
            "{titre} {description}, {respo}",
            "replanter excalibure rendre l'épee à son caillou, dieu",
        ),
    ],
)
async def test_content_with_template(
    client: AsyncClient,
    default_user_headers: dict,
    session: Session,
    default_public_project: Project,
    template: str,
    expected: str,
):
    # Create a slot starting in the past
    # no sms should be issued
    slot_template = SlotTemplate(
        project_id=default_public_project.id,
        title="replanter excalibure",
        description="rendre l'épee à son caillou",
        responsible_contact="dieu",
        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()

    # créer un sms simple
    response = await client.post(
        app.url_path_for("create_sms_batch", project_id=default_public_project.id),
        headers=default_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