|
|
@@ -0,0 +1,343 @@
|
|
|
+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
|