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

import uuid
from httpx import AsyncClient
from sqlalchemy import false, select
from sqlalchemy.orm import Session

from app.main import app
from app.models import Project, Slot, Sms, User, Volunteer
from app.tests.conftest import (
    default_project_id,
    default_volunteer_id,
    default_slot_id,
)


async def test_read_list_project_volunteers(
    client: AsyncClient, default_user_headers: dict, default_public_project: Project
):
    response = await client.get(
        app.url_path_for("list_project_volunteers", project_id=default_project_id),
    )
    assert response.status_code == 401
    response = await client.get(
        app.url_path_for("list_project_volunteers", project_id=uuid.uuid4()),
        headers=default_user_headers,
    )
    assert response.status_code == 404
    response = await client.get(
        app.url_path_for("list_project_volunteers", project_id="pas un uuid valid"),
        headers=default_user_headers,
    )
    assert response.status_code == 422

    response = await client.get(
        app.url_path_for("list_project_volunteers", project_id=default_project_id),
        headers=default_user_headers,
    )
    assert response.status_code == 200
    data = response.json()
    assert len(data) == 1
    volunteer_response = data[0]
    assert volunteer_response["name"] == "Arthur"
    assert volunteer_response["id"] == default_volunteer_id
    assert "created_at" in volunteer_response


async def test_create_volunteer(
    client: AsyncClient,
    default_public_project: Project,
    default_user_headers: dict,
    session: Session,
):
    # Test without autentication
    response = await client.post(
        app.url_path_for("create_volunteer", project_id=default_project_id)
    )
    assert response.status_code == 401
    payload = {
        "name": "Lancelot",
        "email": "lancelot@dulac.fr",
        "phone_number": "03 14 15 92 65",
    }
    # test invalid project_id
    response = await client.post(
        app.url_path_for("create_volunteer", project_id=uuid.uuid4()),
        json=payload,
        headers=default_user_headers,
    )
    assert response.status_code == 404

    # Test normal payload
    response = await client.post(
        app.url_path_for("create_volunteer", project_id=default_project_id),
        json=payload,
        headers=default_user_headers,
    )
    assert response.status_code == 200
    assert response.json()["id"] != default_project_id
    assert response.json()["name"] == "Lancelot"
    assert response.json()["comment"] == ""
    result = session.execute(select(Volunteer).where(Volunteer.project_id == default_project_id))
    volunteers = result.scalars().all()
    assert len(volunteers) > 1

    # test invalid payload
    del payload["name"]
    response = await client.post(
        app.url_path_for("create_volunteer", project_id=default_project_id),
        json=payload,
        headers=default_user_headers,
    )
    assert response.status_code == 422


async def test_create_volunteer_comment(
    client: AsyncClient,
    default_public_project: Project,
    default_user_headers: dict,
    session: Session,
):
    payload = {
        "name": "Lancelot",
        "email": "lancelot@dulac.fr",
        "phone_number": "03 14 15 92 65",
        "comment": "it's a knight",
    }
    # Test normal payload
    response = await client.post(
        app.url_path_for("create_volunteer", project_id=default_project_id),
        json=payload,
        headers=default_user_headers,
    )
    assert response.status_code == 200
    assert response.json()["id"] != default_project_id
    assert response.json()["name"] == "Lancelot"
    assert response.json()["comment"] != ""
    new_id = response.json()["id"]
    result = session.execute(select(Volunteer).where(Volunteer.id == new_id))
    volunteer = result.scalar_one_or_none()
    assert volunteer is not None
    assert volunteer.comment == "it's a knight"


async def test_update_volunteer(
    client: AsyncClient,
    default_public_project: Project,
    default_user_headers: dict,
    session: Session,
):
    # Test without autentication
    response = await client.post(
        app.url_path_for(
            "update_volunteer",
            project_id=default_project_id,
            volunteer_id=default_volunteer_id,
        )
    )
    assert response.status_code == 401

    payload = {
        "name": "Lancelot",
        "email": "lancelot@dulac.fr",
        "phone_number": "03 14 15 92 65",
        "automatic_sms": False,
        "comment": "new comment",
    }

    # test invalid project_id
    response = await client.post(
        app.url_path_for(
            "update_volunteer",
            project_id=uuid.uuid4(),
            volunteer_id=default_volunteer_id,
        ),
        json=payload,
        headers=default_user_headers,
    )
    assert response.status_code == 404

    # test invalid volunteer_id
    response = await client.post(
        app.url_path_for(
            "update_volunteer",
            project_id=default_project_id,
            volunteer_id=uuid.uuid4(),
        ),
        json=payload,
        headers=default_user_headers,
    )
    assert response.status_code == 404

    # Test normal payload

    for k, v in payload.items():
        response = await client.post(
            app.url_path_for(
                "update_volunteer",
                project_id=default_project_id,
                volunteer_id=default_volunteer_id,
            ),
            json={k: v},
            headers=default_user_headers,
        )
        assert response.status_code == 200
        assert response.json()["id"] == default_volunteer_id
        assert response.json()[k] == v


async def test_update_volunteer_slots(
    client: AsyncClient,
    default_public_project: Project,
    default_user_headers: dict,
    session: Session,
):
    response = await client.post(
        app.url_path_for(
            "update_volunteer",
            project_id=default_project_id,
            volunteer_id=default_volunteer_id,
        ),
        json={"slots": []},
        headers=default_user_headers,
    )

    assert response.status_code == 200
    result = session.execute(select(Slot).where(Slot.id == default_slot_id))
    slot = result.scalars().first()
    assert slot is not None
    assert slot.volunteers_id == []

    response = await client.post(
        app.url_path_for(
            "update_volunteer",
            project_id=default_project_id,
            volunteer_id=default_volunteer_id,
        ),
        json={"slots": [default_slot_id]},
        headers=default_user_headers,
    )

    assert response.status_code == 200
    session.refresh(slot)
    assert slot is not None
    assert slot.volunteers_id == [default_volunteer_id]

    # An invalid slot list
    response = await client.post(
        app.url_path_for(
            "update_volunteer",
            project_id=default_project_id,
            volunteer_id=default_volunteer_id,
        ),
        json={"slots": [str(uuid.uuid4())]},
        headers=default_user_headers,
    )

    assert response.status_code == 400

    # An invalid slot list
    response = await client.post(
        app.url_path_for(
            "update_volunteer",
            project_id=default_project_id,
            volunteer_id=default_volunteer_id,
        ),
        json={"slots": ["not uuid str"]},
        headers=default_user_headers,
    )

    assert response.status_code == 422


async def test_delete_volunteer(
    client: AsyncClient,
    default_user_headers: dict,
    session: Session,
    default_public_project: Project,
):
    # Fail deleting the project due to not logged in
    response = await client.delete(
        app.url_path_for(
            "delete_volunteer",
            project_id=default_project_id,
            volunteer_id=default_volunteer_id,
        )
    )
    assert response.status_code == 401
    result = session.execute(select(Volunteer).where(Volunteer.id == default_volunteer_id))
    volunteer = result.scalars().first()
    assert volunteer is not None

    # Proper deletion
    response = await client.delete(
        app.url_path_for(
            "delete_volunteer",
            project_id=default_project_id,
            volunteer_id=default_volunteer_id,
        ),
        headers=default_user_headers,
    )
    assert response.status_code == 200
    result = session.execute(select(Volunteer).where(Volunteer.id == default_volunteer_id))
    volunteer = result.scalars().first()
    assert volunteer is None

    # check deletion is cascaded to slots
    result = session.execute(select(Slot).where(Slot.id == default_slot_id))
    slot: Slot | None = result.scalar_one_or_none()
    assert slot is not None
    assert default_volunteer_id not in slot.volunteers_id

    # Idempotence test
    response = await client.delete(
        app.url_path_for(
            "delete_volunteer",
            project_id=default_project_id,
            volunteer_id=default_volunteer_id,
        ),
        headers=default_user_headers,
    )
    assert response.status_code == 200


async def test_delete_volunteer_any_uuid(
    client: AsyncClient,
    default_user_headers: dict,
    session: Session,
    default_public_project: Project,
):
    # can delete random uuid
    response = await client.delete(
        app.url_path_for(
            "delete_volunteer", project_id=default_project_id, volunteer_id=uuid.uuid4()
        ),
        headers=default_user_headers,
    )
    assert response.status_code == 200

    # Cannot delete non uuid string
    response = await client.delete(
        app.url_path_for(
            "delete_volunteer", project_id=default_project_id, volunteer_id="not uidstr"
        ),
        headers=default_user_headers,
    )
    assert response.status_code == 422


async def test_delete_volunteer_with_sms(
    client: AsyncClient,
    default_user_headers: dict,
    session: Session,
    default_public_project: Project,
):
    sms = Sms()
    sms.project_id = default_project_id
    sms.content = "coucou"
    sms.phone_number = "02 66 66 66 66 66"
    sms.volunteer_id = default_volunteer_id
    session.add(sms)
    session.commit()
    session.refresh(sms)

    response = await client.delete(
        app.url_path_for(
            "delete_volunteer",
            project_id=default_project_id,
            volunteer_id=default_volunteer_id,
        ),
        headers=default_user_headers,
    )
    assert response.status_code == 200
    # Volunteer must be deleted
    result = session.execute(select(Volunteer).where(Volunteer.id == default_volunteer_id))
    assert result.scalar_one_or_none() is None

    # Sms must be deleted
    assert session.execute(select(Sms).where(Sms.id == sms.id)).scalar_one_or_none() is None