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

import uuid

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

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


class TestListVolunteer:
    async def test_read_list_not_authenticated(
        self,
        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

    @pytest.mark.parametrize(
        "project_id,status_code",
        [(uuid.uuid4(), 404), ("pas un uuid valid", 422)],
    )
    async def test_read_list_fails(
        self,
        client: AsyncClient,
        default_user_headers: dict,
        default_public_project: Project,
        make_org_user,
        project_id,
        status_code,
    ):
        _, headers = make_org_user()
        response = await client.get(
            app.url_path_for("list_project_volunteers", project_id=project_id),
            headers=headers,
        )
        assert response.status_code == status_code

    async def test_real_project(
        self, 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),
            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


class TestCreateVolunteer:
    async def test_create_volunteer_without_auth(
        self,
        client: AsyncClient,
    ):
        # Test without authentication
        response = await client.post(
            app.url_path_for("create_volunteer", project_id=default_project_id)
        )
        assert response.status_code == 401

    async def test_create_volunteer_invalid_project_id(
        self,
        client: AsyncClient,
        default_user_headers: dict,
    ):
        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

    async def test_create_volunteer_invalid_payload(
        self,
        client: AsyncClient,
        default_user_headers: dict,
    ):
        payload = {
            "email": "lancelot@dulac.fr",
            "phone_number": "03 14 15 92 65",
        }
        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_valid_payload(
        self,
        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",
        }
        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

    async def test_create_volunteer_with_slots(
        self,
        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",
            "slots": [default_slot_id],
        }
        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
        result = session.execute(
            select(Volunteer).where(Volunteer.project_id == default_project_id)
        )
        volunteers = result.scalars().all()
        assert len(volunteers) > 1
        assert len(volunteers[0].slots) == 1


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