bdlg2023-sms-back/app/tests/test_commissions.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 Commission, CommissionMember, Organization, OrgRole, Project, User
from app.tests.shared_access import SharedProjectAccessTests

pytestmark = pytest.mark.asyncio

MANAGE_ROLES = [OrgRole.ORG_ADMIN, OrgRole.RESPO_BENEVOLE]
FORBIDDEN_MANAGE_ROLES = [OrgRole.RESPO_COMMISSION]
READ_ROLES = [OrgRole.ORG_ADMIN, OrgRole.RESPO_BENEVOLE, OrgRole.RESPO_COMMISSION]


@pytest.fixture
def two_users() -> tuple[User, User]:
    with session_maker() as db:
        u1 = User(
            email="u1@test.com",
            hashed_password="hashed1",
            name="Arthur Pendragon",
            phone_number="0601010101",
        )
        u2 = User(
            email="u2@test.com",
            hashed_password="hashed2",
            name="Merlin Enchanteur",
            phone_number="0602020202",
        )
        db.add_all([u1, u2])
        db.commit()
        db.refresh(u1)
        db.refresh(u2)
        return u1, u2


@pytest.fixture
def default_commission(default_project: Project) -> Commission:
    with session_maker() as db:
        commission = Commission(project_id=default_project.id, name="Logistique")
        db.add(commission)
        db.commit()
        db.refresh(commission)
        return commission


COMMISSION_ROUTES = [
    ("GET", "list_project_commissions", {"project_id": "PROJECT"}, None),
    ("POST", "create_commission", {"project_id": "PROJECT"}, {"name": "Test"}),
    ("GET", "get_commission", {"project_id": "PROJECT", "commission_id": "COMMISSION"}, None),
    (
        "PATCH",
        "update_commission",
        {"project_id": "PROJECT", "commission_id": "COMMISSION"},
        {"name": "Updated"},
    ),
    ("DELETE", "delete_commission", {"project_id": "PROJECT", "commission_id": "COMMISSION"}, None),
    (
        "POST",
        "add_members_to_commission",
        {"project_id": "PROJECT", "commission_id": "COMMISSION"},
        {"user_ids": ["USER"]},
    ),
    (
        "DELETE",
        "remove_member_from_commission",
        {
            "project_id": "PROJECT",
            "commission_id": "COMMISSION",
            "user_id": "USER",
        },
        None,
    ),
]


class TestCommissionCrossCutting(SharedProjectAccessTests):
    @pytest.fixture(params=COMMISSION_ROUTES, ids=lambda x: f"{x[0]}-{x[1]}")
    def resolved_route(self, request, default_project, default_commission):
        method, endpoint, kwargs, payload = request.param
        resolved_kwargs = {
            k: (
                default_project.id
                if v == "PROJECT"
                else default_commission.id
                if v == "COMMISSION"
                else v
            )
            for k, v in kwargs.items()
        }

        return method, endpoint, resolved_kwargs, payload


class TestListCommissions:
    @pytest.mark.parametrize("role", READ_ROLES)
    async def test_role_access(
        self,
        client: AsyncClient,
        default_project: Project,
        default_commission: Commission,
        make_org_user,
        role,
    ):
        _, headers = make_org_user(role=role)
        response = await client.get(
            app.url_path_for("list_project_commissions", project_id=default_project.id),
            headers=headers,
        )
        assert response.status_code == 200
        data = response.json()
        assert len(data) == 1
        assert data[0]["name"] == "Logistique"


class TestCreateCommission:
    @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_commission", project_id=default_project.id),
            headers=headers,
            json={"name": "Accueil"},
        )
        assert response.status_code == 200
        commission = (
            session.execute(select(Commission).where(Commission.name == "Accueil"))
            .scalars()
            .first()
        )
        assert commission is not None
        assert commission.project_id == str(default_project.id)

    @pytest.mark.parametrize("role", FORBIDDEN_MANAGE_ROLES)
    async def test_forbidden_manage_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_commission", project_id=default_project.id),
            headers=headers,
            json={"name": "Nope"},
        )
        assert response.status_code == 403


class TestGetCommission:
    async def test_commission_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_commission", project_id=default_project.id, commission_id=uuid.uuid4()
            ),
            headers=headers,
        )
        assert response.status_code == 404

    async def test_commission_from_other_project_not_found(
        self, client: AsyncClient, default_commission: Commission, 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", is_public=False, organization_id=other_org.id)
        session.add(other_project)
        session.commit()

        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        response = await client.get(
            app.url_path_for(
                "get_commission", project_id=other_project.id, commission_id=default_commission.id
            ),
            headers=headers,
        )
        # Even if they have the org admin role, the project_id URL mismatch means 403 or 404
        assert response.status_code in (403, 404)


class TestUpdateCommission:
    async def test_updates_name(
        self,
        client: AsyncClient,
        default_project: Project,
        default_commission: Commission,
        make_org_user,
        session: Session,
    ):
        _, headers = make_org_user(role=OrgRole.RESPO_BENEVOLE)
        response = await client.patch(
            app.url_path_for(
                "update_commission",
                project_id=default_project.id,
                commission_id=default_commission.id,
            ),
            headers=headers,
            json={"name": "Securite Renamed"},
        )
        assert response.status_code == 200
        commission = session.get(Commission, default_commission.id)
        assert commission.name == "Securite Renamed"


class TestDeleteCommission:
    async def test_deletes_commission_and_members_links_but_not_users(
        self,
        client: AsyncClient,
        default_project: Project,
        default_commission: Commission,
        two_users,
        make_org_user,
        session: Session,
    ):
        u1, _ = two_users
        commission = session.get(Commission, default_commission.id)
        commission.members.append(CommissionMember(user_id=u1.id))
        session.commit()

        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        response = await client.delete(
            app.url_path_for(
                "delete_commission",
                project_id=default_project.id,
                commission_id=default_commission.id,
            ),
            headers=headers,
        )

        assert response.status_code == 200
        assert session.get(Commission, default_commission.id) is None
        # User should still exist
        assert session.get(User, u1.id) is not None
        # Association row should be cascaded
        assoc = (
            session.execute(
                select(CommissionMember).where(
                    CommissionMember.commission_id == default_commission.id
                )
            )
            .scalars()
            .all()
        )
        assert len(assoc) == 0


class TestCommissionMembership:
    async def test_add_users_to_commission(
        self,
        client: AsyncClient,
        default_project: Project,
        default_commission: Commission,
        two_users,
        make_org_user,
    ):
        u1, u2 = two_users
        _, headers = make_org_user(role=OrgRole.RESPO_BENEVOLE)
        response = await client.post(
            app.url_path_for(
                "add_members_to_commission",
                project_id=default_project.id,
                commission_id=default_commission.id,
            ),
            headers=headers,
            json={"user_ids": [str(u1.id), str(u2.id)]},
        )
        assert response.status_code == 200
        data = response.json()
        assert len(data["contacts"]) == 2
        names = [contact["name"] for contact in data["contacts"]]
        assert "Arthur Pendragon" in names
        assert "Merlin Enchanteur" in names

    async def test_add_is_idempotent(
        self,
        client: AsyncClient,
        default_project: Project,
        default_commission: Commission,
        two_users,
        make_org_user,
    ):
        u1, _ = two_users
        _, headers = make_org_user(role=OrgRole.RESPO_BENEVOLE)

        await client.post(
            app.url_path_for(
                "add_members_to_commission",
                project_id=default_project.id,
                commission_id=default_commission.id,
            ),
            headers=headers,
            json={"user_ids": [str(u1.id)]},
        )

        response = await client.post(
            app.url_path_for(
                "add_members_to_commission",
                project_id=default_project.id,
                commission_id=default_commission.id,
            ),
            headers=headers,
            json={"user_ids": [str(u1.id)]},
        )
        assert response.status_code == 200
        assert len(response.json()["contacts"]) == 1

    async def test_remove_member_from_commission(
        self,
        client: AsyncClient,
        default_project: Project,
        default_commission: Commission,
        two_users,
        make_org_user,
        session: Session,
    ):
        u1, u2 = two_users
        commission = session.get(Commission, default_commission.id)
        commission.members.append(CommissionMember(user_id=u1.id))
        commission.members.append(CommissionMember(user_id=u2.id))
        session.commit()

        _, headers = make_org_user(role=OrgRole.RESPO_BENEVOLE)
        response = await client.delete(
            app.url_path_for(
                "remove_member_from_commission",
                project_id=default_project.id,
                commission_id=default_commission.id,
                user_id=u1.id,
            ),
            headers=headers,
        )

        assert response.status_code == 200
        data = response.json()
        assert len(data["contacts"]) == 1
        assert data["contacts"][0]["name"] == u2.name