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

import uuid

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

from app.core.session import session as session_maker
from app.main import app
from app.models import Organization, OrgRole, Project, Volunteer, VolunteerGroup
from app.tests.conftest import default_project_id, default_slot_id
from app.tests.shared_access import SharedProjectAccessTests

pytestmark = pytest.mark.asyncio

MANAGE_ROLES = [OrgRole.ORG_ADMIN, OrgRole.RESPO_BENEVOLE]
FORBIDDEN_ROLES = [OrgRole.RESPO_COMMISSION, None]


@pytest.fixture
def two_volunteers(default_project: Project) -> tuple[Volunteer, Volunteer]:
    with session_maker() as db:
        v1 = Volunteer(
            project_id=default_project.id,
            name="Yvain",
            surname="Lion",
            email="y@test.com",
            phone_number="0601010101",
            automatic_sms=True,
        )
        v2 = Volunteer(
            project_id=default_project.id,
            name="Lancelot",
            surname="Lac",
            email="l@test.com",
            phone_number="0602020202",
            automatic_sms=True,
        )
        db.add_all([v1, v2])
        db.commit()
        db.refresh(v1)
        db.refresh(v2)
        return v1, v2


@pytest.fixture
def default_group(default_project: Project) -> VolunteerGroup:
    with session_maker() as db:
        group = VolunteerGroup(project_id=default_project.id, name="Chevaliers", color="#FF0000")
        db.add(group)
        db.commit()
        db.refresh(group)
        return group


route_kwargs = {"project_id": default_project_id}
route_kwargs_2 = {**route_kwargs, "group_id": "GROUP"}
VOLUNTEER_GROUP_ROUTES = [
    ("GET", "list_project_groups", route_kwargs, None),
    ("POST", "create_group", route_kwargs, {"name": "Test"}),
    ("GET", "get_group", route_kwargs_2, None),
    ("POST", "update_group", route_kwargs_2, {"name": "Updated"}),
    ("DELETE", "delete_group", route_kwargs_2, None),
    ("POST", "add_volunteers_to_group", route_kwargs_2, {"volunteer_ids": []}),
    ("DELETE", "remove_volunteer_from_group", {**route_kwargs_2, "volunteer_id": "VOL"}, None),
    ("POST", "send_sms_to_group", route_kwargs_2, {"content": "coucou"}),
]


class TestVolunteerGroupCrossCuttingAccess(SharedProjectAccessTests):
    @pytest.fixture(params=VOLUNTEER_GROUP_ROUTES, ids=lambda x: f"{x[0]}-{x[1]}")
    def resolved_route(self, request, default_group, two_volunteers):
        method, endpoint, kwargs, payload = request.param
        resolved_kwargs = {
            k: (
                default_group.id
                if v == "GROUP"
                else two_volunteers[0].id
                if v == "VOL"
                else default_slot_id
                if v == "SLOT"
                else v
            )
            for k, v in kwargs.items()
        }
        return method, endpoint, resolved_kwargs, payload


class TestListGroups:
    @pytest.mark.parametrize("role", MANAGE_ROLES)
    async def test_role_access(
        self,
        client: AsyncClient,
        default_project: Project,
        default_group: VolunteerGroup,
        make_org_user,
        role,
    ):
        _, headers = make_org_user(role=role)
        response = await client.get(
            app.url_path_for("list_project_groups", project_id=default_project.id), headers=headers
        )
        assert response.status_code == 200
        data = response.json()
        assert len(data) == 1
        assert data[0]["name"] == "Chevaliers"

    @pytest.mark.parametrize("role", FORBIDDEN_ROLES)
    async def test_forbidden_roles(
        self, client: AsyncClient, default_project: Project, make_org_user, role
    ):
        _, headers = make_org_user(role=role)
        response = await client.get(
            app.url_path_for("list_project_groups", project_id=default_project.id), headers=headers
        )
        assert response.status_code == 403

    async def test_project_not_found(self, client: AsyncClient, make_org_user):
        _, headers = (
            make_org_user(role=OrgRole.ORG_ADMIN, global_role=None)
            if False
            else make_org_user(role=None)
        )
        # super_admin needed since require_org_role 404s on missing project regardless of role
        from app.models import GlobalRole

        _, headers = make_org_user(role=None, global_role=GlobalRole.SUPER_ADMIN)
        response = await client.get(
            app.url_path_for("list_project_groups", project_id=uuid.uuid4()), headers=headers
        )
        assert response.status_code == 404


class TestCreateGroup:
    @pytest.mark.parametrize("role", MANAGE_ROLES)
    async def test_role_access(
        self, client: AsyncClient, default_project: Project, make_org_user, session: Session, role
    ):
        _, headers = make_org_user(role=role)
        response = await client.post(
            app.url_path_for("create_group", project_id=default_project.id),
            headers=headers,
            json={"name": "Bar team", "color": "#00FF00"},
        )
        assert response.status_code == 200
        group = (
            session.execute(select(VolunteerGroup).where(VolunteerGroup.name == "Bar team"))
            .scalars()
            .first()
        )
        assert group is not None
        assert group.color == "#00FF00"

    @pytest.mark.parametrize("role", FORBIDDEN_ROLES)
    async def test_forbidden_roles(
        self, client: AsyncClient, default_project: Project, make_org_user, role
    ):
        _, headers = make_org_user(role=role)
        response = await client.post(
            app.url_path_for("create_group", project_id=default_project.id),
            headers=headers,
            json={"name": "Nope"},
        )
        assert response.status_code == 403


class TestGetGroup:
    async def test_not_found(self, client: AsyncClient, make_org_user, default_project: Project):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        response = await client.get(
            app.url_path_for("get_group", project_id=default_project.id, group_id=uuid.uuid4()),
            headers=headers,
        )
        assert response.status_code == 404

    async def test_group_from_other_project_not_found(
        self, client: AsyncClient, default_group: VolunteerGroup, make_org_user, session: Session
    ):
        """A group belonging to a different project must 404, even for a
        role that would otherwise pass, since group.project_id doesn't
        match the project_id in the URL."""
        other_org = Organization(id=str(uuid.uuid4()), name="Other Org")
        session.add(other_org)
        session.commit()
        other_project = Project(name="Other Project", is_public=False, organization_id=other_org.id)
        session.add(other_project)
        session.commit()

        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)  # member of default_organization
        response = await client.get(
            app.url_path_for("get_group", project_id=other_project.id, group_id=default_group.id),
            headers=headers,
        )
        # 403 (not a member of other_project's org) fires before the group-mismatch
        # check even runs -- both are valid rejections, but confirm which one applies
        assert response.status_code in (403, 404)


class TestUpdateGroup:
    async def test_updates_name_and_color(
        self,
        client: AsyncClient,
        default_project: Project,
        default_group: VolunteerGroup,
        make_org_user,
        session: Session,
    ):
        _, headers = make_org_user(role=OrgRole.RESPO_BENEVOLE)
        response = await client.post(
            app.url_path_for(
                "update_group", project_id=default_project.id, group_id=default_group.id
            ),
            headers=headers,
            json={"name": "Renamed", "color": "#0000FF"},
        )
        assert response.status_code == 200
        group = session.get(VolunteerGroup, default_group.id)
        assert group.name == "Renamed"
        assert group.color == "#0000FF"


class TestDeleteGroup:
    async def test_deletes_group_without_deleting_volunteers(
        self,
        client: AsyncClient,
        default_project: Project,
        default_group: VolunteerGroup,
        two_volunteers,
        make_org_user,
        session: Session,
    ):
        v1, _ = two_volunteers
        group = session.get(VolunteerGroup, default_group.id)
        volunteer = session.get(Volunteer, v1.id)
        group.volunteers.append(volunteer)
        session.commit()

        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        response = await client.delete(
            app.url_path_for(
                "delete_group", project_id=default_project.id, group_id=default_group.id
            ),
            headers=headers,
        )
        assert response.status_code == 200
        assert session.get(VolunteerGroup, default_group.id) is None
        assert session.get(Volunteer, v1.id) is not None


class TestGroupMembership:
    async def test_add_volunteers_to_group(
        self,
        client: AsyncClient,
        default_project: Project,
        default_group: VolunteerGroup,
        two_volunteers,
        make_org_user,
    ):
        v1, v2 = two_volunteers
        _, headers = make_org_user(role=OrgRole.RESPO_BENEVOLE)
        response = await client.post(
            app.url_path_for(
                "add_volunteers_to_group", project_id=default_project.id, group_id=default_group.id
            ),
            headers=headers,
            json={"volunteer_ids": [v1.id, v2.id]},
        )
        assert response.status_code == 200
        assert sorted(response.json()["volunteers_id"]) == sorted([v1.id, v2.id])

    async def test_add_is_idempotent(
        self,
        client: AsyncClient,
        default_project: Project,
        default_group: VolunteerGroup,
        two_volunteers,
        make_org_user,
    ):
        v1, _ = two_volunteers
        _, headers = make_org_user(role=OrgRole.RESPO_BENEVOLE)
        await client.post(
            app.url_path_for(
                "add_volunteers_to_group", project_id=default_project.id, group_id=default_group.id
            ),
            headers=headers,
            json={"volunteer_ids": [v1.id]},
        )
        response = await client.post(
            app.url_path_for(
                "add_volunteers_to_group", project_id=default_project.id, group_id=default_group.id
            ),
            headers=headers,
            json={"volunteer_ids": [v1.id]},
        )
        assert response.status_code == 200
        assert response.json()["volunteers_id"] == [v1.id]

    async def test_volunteer_from_other_project_rejected(
        self,
        client: AsyncClient,
        default_project: Project,
        default_group: VolunteerGroup,
        make_org_user,
        session: Session,
    ):
        other_org = Organization(id=str(uuid.uuid4()), name="Other Org")
        session.add(other_org)
        session.commit()
        other_project = Project(
            name="Other Project 2", is_public=False, organization_id=other_org.id
        )
        session.add(other_project)
        session.commit()
        stray_volunteer = Volunteer(
            project_id=other_project.id,
            name="Stray",
            surname="Cat",
            email="s@test.com",
            phone_number="0600000000",
        )
        session.add(stray_volunteer)
        session.commit()

        _, headers = make_org_user(role=OrgRole.RESPO_BENEVOLE)
        response = await client.post(
            app.url_path_for(
                "add_volunteers_to_group", project_id=default_project.id, group_id=default_group.id
            ),
            headers=headers,
            json={"volunteer_ids": [stray_volunteer.id]},
        )
        assert response.status_code == 400

    async def test_remove_volunteer_from_group(
        self,
        client: AsyncClient,
        default_project: Project,
        default_group: VolunteerGroup,
        two_volunteers,
        make_org_user,
        session: Session,
    ):
        v1, v2 = two_volunteers
        group = session.get(VolunteerGroup, default_group.id)
        group.volunteers.append(session.get(Volunteer, v1.id))
        group.volunteers.append(session.get(Volunteer, v2.id))
        session.commit()

        _, headers = make_org_user(role=OrgRole.RESPO_BENEVOLE)
        response = await client.delete(
            app.url_path_for(
                "remove_volunteer_from_group",
                project_id=default_project.id,
                group_id=default_group.id,
                volunteer_id=v1.id,
            ),
            headers=headers,
        )
        assert response.status_code == 200
        assert response.json()["volunteers_id"] == [v2.id]


class TestSendSmsToGroup:
    async def test_sends_to_each_member_with_automatic_sms(
        self,
        client: AsyncClient,
        default_project: Project,
        default_group: VolunteerGroup,
        two_volunteers,
        make_org_user,
        session: Session,
    ):
        v1, v2 = two_volunteers
        group = session.get(VolunteerGroup, default_group.id)
        group.volunteers.append(session.get(Volunteer, v1.id))
        group.volunteers.append(session.get(Volunteer, v2.id))
        session.commit()

        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        response = await client.post(
            app.url_path_for(
                "send_sms_to_group", project_id=default_project.id, group_id=default_group.id
            ),
            headers=headers,
            json={"content": "Bonjour {prenom} {nom} !"},
        )
        assert response.status_code == 200
        contents = sorted(sms["content"] for sms in response.json())
        assert contents == ["Bonjour Lancelot Lac !", "Bonjour Yvain Lion !"]

    async def test_skips_volunteers_without_automatic_sms(
        self,
        client: AsyncClient,
        default_project: Project,
        default_group: VolunteerGroup,
        two_volunteers,
        make_org_user,
        session: Session,
    ):
        v1, v2 = two_volunteers
        vol2 = session.get(Volunteer, v2.id)
        vol2.automatic_sms = False
        group = session.get(VolunteerGroup, default_group.id)
        group.volunteers.append(session.get(Volunteer, v1.id))
        group.volunteers.append(session.get(Volunteer, v2.id))
        session.commit()

        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        response = await client.post(
            app.url_path_for(
                "send_sms_to_group", project_id=default_project.id, group_id=default_group.id
            ),
            headers=headers,
            json={"content": "Bonjour {prenom} !"},
        )
        assert response.status_code == 200
        assert len(response.json()) == 1
        assert response.json()[0]["content"] == "Bonjour Yvain !"