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, Organization, OrgRole, Project, User, UserOrganization
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


route_kwarg = {"project_id": "PROJECT", "commission_id": "COMMISSION"}
COMMISSION_ROUTES = [
    ("GET", "list_project_commissions", {"project_id": "PROJECT"}, None),
    ("POST", "create_commission", {"project_id": "PROJECT"}, {"name": "Test"}),
    ("GET", "get_commission", route_kwarg, None),
    ("POST", "update_commission", route_kwarg, {"name": "Updated"}),
    ("DELETE", "delete_commission", route_kwarg, None),
    ("POST", "add_members_to_commission", route_kwarg, {"user_ids": ["USER"]}),
    ("DELETE", "remove_member_from_commission", {**route_kwarg, "user_id": "USER"}, None),
    (
        "POST",
        "invite_commission_member",
        route_kwarg,
        {"email": "crosscut@test.com", "name": "Cross Cut", "password": "x"},
    ),
]


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.post(
            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(u1)
        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


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["members_id"]) == 2
        assert str(u1.id) in data["members_id"]
        assert str(u2.id) in data["members_id"]

    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()["members_id"]) == 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)
        assert commission is not None
        commission.members.append(u1)
        commission.members.append(u2)
        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["members_id"]) == 1
        assert data["members_id"][0] == str(u2.id)


class TestInviteCommissionMember:
    @pytest.mark.parametrize("role", MANAGE_ROLES)
    async def test_creates_account_and_adds_to_commission(
        self,
        client: AsyncClient,
        default_project: Project,
        default_commission: Commission,
        make_org_user,
        session: Session,
        role,
    ):
        _, headers = make_org_user(role=role)
        response = await client.post(
            app.url_path_for(
                "invite_commission_member",
                project_id=default_project.id,
                commission_id=default_commission.id,
            ),
            headers=headers,
            json={
                "email": "lancelot@test.com",
                "name": "Lancelot du Lac",
                "password": "TempPass123!",
                "org_role": "respo_commission",
            },
        )
        assert response.status_code == 200
        data = response.json()

        user = session.scalar(select(User).where(User.email == "lancelot@test.com"))
        assert user is not None
        assert user.name == "Lancelot du Lac"
        assert user.must_change_password is True
        assert str(user.id) in data["members_id"]

        membership = session.get(UserOrganization, (user.id, default_project.organization_id))
        assert membership is not None
        assert membership.role == OrgRole.RESPO_COMMISSION

    @pytest.mark.parametrize("role", FORBIDDEN_MANAGE_ROLES)
    async def test_forbidden_manage_roles(
        self,
        client: AsyncClient,
        default_project: Project,
        default_commission: Commission,
        make_org_user,
        role,
    ):
        _, headers = make_org_user(role=role)
        response = await client.post(
            app.url_path_for(
                "invite_commission_member",
                project_id=default_project.id,
                commission_id=default_commission.id,
            ),
            headers=headers,
            json={"email": "nope@test.com", "name": "Nope", "password": "x"},
        )
        assert response.status_code == 403

    async def test_existing_user_not_recreated_and_password_untouched(
        self,
        client: AsyncClient,
        default_project: Project,
        default_commission: Commission,
        two_users,
        make_org_user,
        session: Session,
    ):
        u1, _ = two_users
        original_hash = u1.hashed_password

        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        response = await client.post(
            app.url_path_for(
                "invite_commission_member",
                project_id=default_project.id,
                commission_id=default_commission.id,
            ),
            headers=headers,
            json={
                "email": u1.email,
                "name": "Ignored Name Change",
                "password": "AttackerChosenPassword!",
                "org_role": "respo_commission",
            },
        )
        assert response.status_code == 200

        refreshed = session.get(User, u1.id)
        assert refreshed.hashed_password == original_hash  # untouched
        assert refreshed.name == "Arthur Pendragon"  # untouched
        assert refreshed.must_change_password is False  # never set on existing account
        assert str(u1.id) in response.json()["members_id"]

    async def test_existing_org_membership_role_not_overwritten(
        self,
        client: AsyncClient,
        default_project: Project,
        default_commission: Commission,
        two_users,
        make_org_user,
        session: Session,
    ):
        u1, _ = two_users
        session.add(
            UserOrganization(
                user_id=u1.id,
                organization_id=default_project.organization_id,
                role=OrgRole.RESPO_BENEVOLE,
            )
        )
        session.commit()

        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        response = await client.post(
            app.url_path_for(
                "invite_commission_member",
                project_id=default_project.id,
                commission_id=default_commission.id,
            ),
            headers=headers,
            # requesting respo_commission, but they're already respo_benevole -- must not downgrade
            json={
                "email": u1.email,
                "name": u1.name,
                "password": "x",
                "org_role": "respo_commission",
            },
        )
        assert response.status_code == 200

        membership = session.get(UserOrganization, (u1.id, default_project.organization_id))
        assert membership.role == OrgRole.RESPO_BENEVOLE  # unchanged

    async def test_respo_benevole_cannot_grant_org_admin(
        self,
        client: AsyncClient,
        default_project: Project,
        default_commission: Commission,
        make_org_user,
    ):
        _, headers = make_org_user(role=OrgRole.RESPO_BENEVOLE)
        response = await client.post(
            app.url_path_for(
                "invite_commission_member",
                project_id=default_project.id,
                commission_id=default_commission.id,
            ),
            headers=headers,
            json={
                "email": "wouldbe@test.com",
                "name": "Would Be Admin",
                "password": "x",
                "org_role": "org_admin",
            },
        )
        assert response.status_code == 403

    async def test_org_admin_can_grant_org_admin(
        self,
        client: AsyncClient,
        default_project: Project,
        default_commission: Commission,
        make_org_user,
        session: Session,
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        response = await client.post(
            app.url_path_for(
                "invite_commission_member",
                project_id=default_project.id,
                commission_id=default_commission.id,
            ),
            headers=headers,
            json={
                "email": "newadmin@test.com",
                "name": "New Admin",
                "password": "x",
                "org_role": "org_admin",
            },
        )
        assert response.status_code == 200
        user = session.scalar(select(User).where(User.email == "newadmin@test.com"))
        membership = session.get(UserOrganization, (user.id, default_project.organization_id))
        assert membership.role == OrgRole.ORG_ADMIN

    async def test_invite_is_idempotent_on_commission_membership(
        self,
        client: AsyncClient,
        default_project: Project,
        default_commission: Commission,
        make_org_user,
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        payload = {
            "email": "guenievre@test.com",
            "name": "Guenièvre",
            "password": "x",
            "org_role": "respo_commission",
        }
        for _ in range(2):
            response = await client.post(
                app.url_path_for(
                    "invite_commission_member",
                    project_id=default_project.id,
                    commission_id=default_commission.id,
                ),
                headers=headers,
                json=payload,
            )
            assert response.status_code == 200
        assert len(response.json()["members_id"]) == 1