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

import uuid
from datetime import datetime, timedelta

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, Slot, Volunteer, VolunteerGroup

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


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.patch(
            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 TestAddGroupToSlot:
    async def test_bulk_assigns_all_group_members(
        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))
        slot = Slot(
            project_id=default_project.id,
            title="Garde du Graal",
            starting_time=datetime.now() + timedelta(hours=1),
            ending_time=datetime.now() + timedelta(hours=2),
        )
        session.add(slot)
        session.commit()

        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        response = await client.post(
            app.url_path_for(
                "add_group_to_slot",
                project_id=default_project.id,
                group_id=default_group.id,
                slot_id=slot.id,
            ),
            headers=headers,
        )
        assert response.status_code == 200
        ids = [v["id"] for v in response.json()]
        assert sorted(ids) == sorted([v1.id, v2.id])

    async def test_slot_from_other_project_not_found(
        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 3", is_public=False, organization_id=other_org.id
        )
        session.add(other_project)
        session.commit()
        stray_slot = Slot(
            project_id=other_project.id,
            title="Stray slot",
            starting_time=datetime.now(),
            ending_time=datetime.now() + timedelta(hours=1),
        )
        session.add(stray_slot)
        session.commit()

        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        response = await client.post(
            app.url_path_for(
                "add_group_to_slot",
                project_id=default_project.id,
                group_id=default_group.id,
                slot_id=stray_slot.id,
            ),
            headers=headers,
        )
        assert response.status_code == 404


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 !"