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.main import app
from app.models import OrgRole, Organization, Project, Slot, Volunteer, VolunteerGroup
from app.core.session import session as session_maker

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