bdlg2023-sms-back/app/tests/test_volunteers.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 OrgRole, Project, Slot, Sms, Volunteer
from app.tests.conftest import default_project_id, default_slot_id, default_volunteer_id
from app.tests.shared_access import SharedProjectAccessTests

pytestmark = pytest.mark.asyncio

WRITE_FORBIDDEN_ROLES = [OrgRole.RESPO_COMMISSION, None]


route_kwargs = {"project_id": default_project_id}
route_kwargs_2 = {**route_kwargs, "volunteer_id": default_volunteer_id}
VOLUNTEER_ROUTES = [
    ("GET", "list_project_volunteers", route_kwargs, None),
    ("POST", "create_volunteer", route_kwargs, {"name": "Test", "email": "a@free.fr"}),
    ("POST", "update_volunteer", route_kwargs_2, {"name": "Updated"}),
    ("DELETE", "delete_volunteer", route_kwargs_2, None),
]


class TestVolunteerCrossCuttingAccess(SharedProjectAccessTests):
    @pytest.fixture(params=VOLUNTEER_ROUTES, ids=lambda x: f"{x[0]}-{x[1]}")
    def resolved_route(self, request, default_project):
        return request.param


class TestListVolunteer:
    @pytest.mark.parametrize(
        "role", [OrgRole.ORG_ADMIN, OrgRole.RESPO_BENEVOLE, OrgRole.RESPO_COMMISSION]
    )
    async def test_readable_by_all_org_roles(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
        role,
    ):
        """respo_commission can read volunteers -- needs visibility into who
        they'll be working with -- even though they can't manage them."""
        _, headers = make_org_user(role=role)
        response = await client.get(
            app.url_path_for("list_project_volunteers", project_id=default_project_id),
            headers=headers,
        )
        assert response.status_code == 200
        data = response.json()
        assert len(data) == 1
        assert data[0]["name"] == "Arthur"
        assert data[0]["id"] == default_volunteer_id
        assert "created_at" in data[0]

    async def test_no_membership_forbidden(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
    ):
        _, headers = make_org_user(role=None)
        response = await client.get(
            app.url_path_for("list_project_volunteers", project_id=default_project_id),
            headers=headers,
        )
        assert response.status_code == 403


class TestCreateVolunteer:
    async def test_invalid_payload(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        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=headers,
        )
        assert response.status_code == 422

    @pytest.mark.parametrize("role", [OrgRole.ORG_ADMIN, OrgRole.RESPO_BENEVOLE])
    async def test_write_roles_create(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
        session: Session,
        role,
    ):
        _, headers = make_org_user(role=role)
        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=headers,
        )
        assert response.status_code == 200
        assert response.json()["id"] != default_project_id
        assert response.json()["name"] == "Lancelot"
        assert response.json()["comment"] == ""

        volunteers = (
            session.execute(select(Volunteer).where(Volunteer.project_id == default_project_id))
            .scalars()
            .all()
        )
        assert len(volunteers) > 1

    @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)
        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=headers,
        )
        assert response.status_code == 403

    async def test_create_with_slots(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
        session: Session,
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        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=headers,
        )
        assert response.status_code == 200
        volunteers = (
            session.execute(select(Volunteer).where(Volunteer.project_id == default_project_id))
            .scalars()
            .all()
        )
        new_volunteer = [v for v in volunteers if v.id != default_volunteer_id][0]
        assert len(new_volunteer.slots) == 1

    async def test_create_with_comment(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
        session: Session,
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        payload = {
            "name": "Lancelot",
            "email": "lancelot@dulac.fr",
            "phone_number": "03 14 15 92 65",
            "comment": "it's a knight",
        }
        response = await client.post(
            app.url_path_for("create_volunteer", project_id=default_project_id),
            json=payload,
            headers=headers,
        )
        assert response.status_code == 200
        assert response.json()["comment"] != ""
        volunteer = session.get(Volunteer, response.json()["id"])
        assert volunteer.comment == "it's a knight"


class TestUpdateVolunteer:
    async def test_invalid_volunteer_id(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        payload = {"name": "Lancelot", "email": "l@dulac.fr", "phone_number": "0314159265"}
        response = await client.post(
            app.url_path_for(
                "update_volunteer", project_id=default_project_id, volunteer_id=uuid.uuid4()
            ),
            json=payload,
            headers=headers,
        )
        assert response.status_code == 404

    @pytest.mark.parametrize("role", [OrgRole.ORG_ADMIN, OrgRole.RESPO_BENEVOLE])
    async def test_write_roles_update_each_field(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
        role,
    ):
        _, headers = make_org_user(role=role)
        payload = {
            "name": "Lancelot",
            "email": "lancelot@dulac.fr",
            "phone_number": "03 14 15 92 65",
            "automatic_sms": False,
            "comment": "new comment",
        }
        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=headers,
            )
            assert response.status_code == 200
            assert response.json()["id"] == default_volunteer_id
            assert response.json()[k] == v

    @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(
                "update_volunteer", project_id=default_project_id, volunteer_id=default_volunteer_id
            ),
            json={"comment": "hijacked"},
            headers=headers,
        )
        assert response.status_code == 403

    async def test_update_slots(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
        session: Session,
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        url = app.url_path_for(
            "update_volunteer", project_id=default_project_id, volunteer_id=default_volunteer_id
        )

        response = await client.post(url, json={"slots": []}, headers=headers)
        assert response.status_code == 200
        slot = session.execute(select(Slot).where(Slot.id == default_slot_id)).scalars().first()
        assert slot.volunteers_id == []

        response = await client.post(url, json={"slots": [default_slot_id]}, headers=headers)
        assert response.status_code == 200
        session.refresh(slot)
        assert slot.volunteers_id == [default_volunteer_id]

        response = await client.post(url, json={"slots": [str(uuid.uuid4())]}, headers=headers)
        assert response.status_code == 400

        response = await client.post(url, json={"slots": ["not uuid str"]}, headers=headers)
        assert response.status_code == 422


class TestDeleteVolunteer:
    @pytest.mark.parametrize("role", WRITE_FORBIDDEN_ROLES)
    async def test_read_only_roles_forbidden(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
        session: Session,
        role,
    ):
        _, headers = make_org_user(role=role)
        response = await client.delete(
            app.url_path_for(
                "delete_volunteer", project_id=default_project_id, volunteer_id=default_volunteer_id
            ),
            headers=headers,
        )
        assert response.status_code == 403
        assert session.get(Volunteer, default_volunteer_id) is not None

    async def test_org_admin_deletes(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
        session: Session,
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        response = await client.delete(
            app.url_path_for(
                "delete_volunteer", project_id=default_project_id, volunteer_id=default_volunteer_id
            ),
            headers=headers,
        )
        assert response.status_code == 200
        assert session.get(Volunteer, default_volunteer_id) is None

        slot = session.execute(select(Slot).where(Slot.id == default_slot_id)).scalars().first()
        assert default_volunteer_id not in slot.volunteers_id

    async def test_idempotent_delete(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        url = app.url_path_for(
            "delete_volunteer", project_id=default_project_id, volunteer_id=default_volunteer_id
        )
        response = await client.delete(url, headers=headers)
        assert response.status_code == 200
        response = await client.delete(url, headers=headers)
        assert response.status_code == 200

    async def test_delete_random_uuid(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
    ):
        """Unlike delete_slot/delete_template, this still bulk-DELETEs
        without an existence check via a Depends() resolver -- so a
        nonexistent volunteer_id stays a silent 200, matching the original
        behavior. If you want 404-on-missing consistency across all three
        routers, this needs the same get()-then-check pattern slots.py uses."""
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        response = await client.delete(
            app.url_path_for(
                "delete_volunteer", project_id=default_project_id, volunteer_id=uuid.uuid4()
            ),
            headers=headers,
        )
        assert response.status_code == 200

    async def test_invalid_uuid_format(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        response = await client.delete(
            app.url_path_for(
                "delete_volunteer", project_id=default_project_id, volunteer_id="not uidstr"
            ),
            headers=headers,
        )
        assert response.status_code == 422

    async def test_delete_cascades_sms(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
        session: Session,
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        sms = Sms(
            project_id=default_project_id,
            content="coucou",
            phone_number="02 66 66 66 66 66",
            volunteer_id=default_volunteer_id,
        )
        session.add(sms)
        session.commit()
        sms_id = sms.id

        response = await client.delete(
            app.url_path_for(
                "delete_volunteer", project_id=default_project_id, volunteer_id=default_volunteer_id
            ),
            headers=headers,
        )
        assert response.status_code == 200
        assert session.get(Volunteer, default_volunteer_id) is None
        assert session.execute(select(Sms).where(Sms.id == sms_id)).scalar_one_or_none() is None