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

import uuid

import pytest
from httpx import AsyncClient
from sqlalchemy import select
from sqlalchemy.orm import Session

from app.main import app
from app.models import GlobalRole, OrgRole, Project, Slot, Volunteer
from app.tests.conftest import default_organization_id, default_project_id, default_project_name
from app.tests.shared_access import SharedProjectAccessTests

pytestmark = pytest.mark.asyncio

ALL_ORG_ROLES = [OrgRole.ORG_ADMIN, OrgRole.RESPO_BENEVOLE, OrgRole.RESPO_COMMISSION]

PROJECT_ROUTES = [
    ("GET", "list_project", {}, None),
    (
        "POST",
        "create_project",
        {},
        {
            "name": "Test",
            "organization_id": default_organization_id,
            "is_public": False,
        },
    ),
    ("POST", "update_project", {"project_id": default_project_id}, {"title": "Updated"}),
    ("POST", "create_sms_batch", {"project_id": default_project_id}, {"template": "Hello"}),
    ("DELETE", "delete_project", {"project_id": default_project_id}, None),
]


class TestProjectCrossCutting(SharedProjectAccessTests):
    @pytest.fixture(params=PROJECT_ROUTES, ids=lambda x: f"{x[0]}-{x[1]}")
    def resolved_route(self, request, default_project):
        return request.param


class TestListProject:
    @pytest.mark.parametrize("role", ALL_ORG_ROLES)
    async def test_org_member_sees_their_project(
        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"), headers=headers)
        assert response.status_code == 200
        data = response.json()
        assert len(data) == 1
        assert data[0]["id"] == default_project_id

    async def test_super_admin_sees_all_orgs(
        self, client: AsyncClient, default_project: Project, make_org_user
    ):
        _, headers = make_org_user(role=None, global_role=GlobalRole.SUPER_ADMIN)
        response = await client.get(app.url_path_for("list_project"), headers=headers)
        assert response.status_code == 200
        assert len(response.json()) == 1

    async def test_excludes_other_orgs_project(
        self,
        client: AsyncClient,
        default_project: Project,
        make_org_user,
        session: Session,
        other_org_user,
    ):
        _, headers = other_org_user
        response = await client.get(app.url_path_for("list_project"), headers=headers)
        assert response.status_code == 200
        assert response.json() == []


class TestGetProject:
    @pytest.mark.parametrize(
        "role, expected_status",
        [
            (OrgRole.ORG_ADMIN, 200),
            (OrgRole.RESPO_BENEVOLE, 200),
            (OrgRole.RESPO_COMMISSION, 200),
            (None, 403),
        ],
    )
    async def test_role_access(
        self, client: AsyncClient, default_project: Project, make_org_user, role, expected_status
    ):
        _, headers = make_org_user(role=role)
        response = await client.get(
            app.url_path_for("get_project", project_id=default_project_id), headers=headers
        )
        assert response.status_code == expected_status
        if expected_status == 200:
            assert response.json()["id"] == default_project_id

    @pytest.mark.parametrize("project_id,status", ((default_project_id, 200), (uuid.uuid4(), 404)))
    async def test_super_admin_bypasses_membership(
        self, client: AsyncClient, default_project: Project, make_org_user, project_id, status
    ):
        _, headers = make_org_user(role=None, global_role=GlobalRole.SUPER_ADMIN)
        response = await client.get(
            app.url_path_for("get_project", project_id=project_id), headers=headers
        )
        assert response.status_code == status


class TestGetPublicProject:
    async def test_missing_or_private_returns_404(
        self, client: AsyncClient, default_project: Project
    ):
        response = await client.get(
            app.url_path_for("get_public_project", project_id=default_project_id)
        )
        assert response.status_code == 404

    async def test_public_project_no_auth_needed(
        self, client: AsyncClient, default_public_project: Project
    ):
        response = await client.get(
            app.url_path_for("get_public_project", project_id=default_project_id)
        )
        assert response.status_code == 200
        assert response.json()["name"] == default_project_name


class TestListPublicProject:
    async def test_excludes_private(self, client: AsyncClient, default_project: Project):
        response = await client.get(app.url_path_for("list_public_project"))
        assert response.status_code == 200
        assert response.json() == []

    async def test_includes_public(self, client: AsyncClient, default_public_project: Project):
        response = await client.get(app.url_path_for("list_public_project"))
        assert response.status_code == 200
        assert len(response.json()) == 1


class TestCreateProject:
    @pytest.mark.parametrize(
        "role, expected_status",
        [
            (OrgRole.ORG_ADMIN, 200),
            (OrgRole.RESPO_BENEVOLE, 403),
            (OrgRole.RESPO_COMMISSION, 403),
            (None, 403),
        ],
    )
    async def test_role_access(self, client: AsyncClient, make_org_user, role, expected_status):
        _, headers = make_org_user(role=role)
        response = await client.post(
            app.url_path_for("create_project"),
            headers=headers,
            json={"name": f"Project-{role}", "organization_id": default_organization_id},
        )
        assert response.status_code == expected_status

    async def test_super_admin_can_create_anywhere(self, client: AsyncClient, make_org_user):
        _, headers = make_org_user(role=None, global_role=GlobalRole.SUPER_ADMIN)
        response = await client.post(
            app.url_path_for("create_project"),
            headers=headers,
            json={"name": "Super Created", "organization_id": default_organization_id},
        )
        assert response.status_code == 200

    async def test_public_flag(self, client: AsyncClient, make_org_user, session: Session):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        response = await client.post(
            app.url_path_for("create_project"),
            headers=headers,
            json={"name": "Public", "is_public": True, "organization_id": default_organization_id},
        )
        print(response.json())
        assert response.status_code == 200
        project = session.execute(select(Project).where(Project.name == "Public")).scalars().first()
        assert project.is_public

    async def test_validation_error(self, client: AsyncClient, make_org_user, session: Session):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        response = await client.post(
            app.url_path_for("create_project"), headers=headers, json={"coucou": "Coucou"}
        )
        assert response.status_code == 422
        assert session.execute(select(Project)).scalars().first() is None

    async def test_existing_name(
        self, client: AsyncClient, make_org_user, default_project: Project, session: Session
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        response = await client.post(
            app.url_path_for("create_project"),
            headers=headers,
            json={"name": default_project_name, "organization_id": default_organization_id},
        )
        assert response.status_code == 400


class TestUpdateProject:
    @pytest.mark.parametrize(
        "role, expected_status",
        [
            (OrgRole.ORG_ADMIN, 200),
            (OrgRole.RESPO_BENEVOLE, 403),
            (OrgRole.RESPO_COMMISSION, 403),
            (None, 403),
        ],
    )
    async def test_role_access(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
        session: Session,
        role,
        expected_status,
    ):
        _, headers = make_org_user(role=role)
        response = await client.post(
            app.url_path_for("update_project", project_id=default_project_id),
            headers=headers,
            json={"name": "Coucou"},
        )
        print(response.json())
        assert response.status_code == expected_status
        if expected_status == 200:
            project = session.get(Project, default_project_id)
            assert project.name == "Coucou"

    async def test_validation_error(
        self, client: AsyncClient, default_public_project: Project, make_org_user
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        response = await client.post(
            app.url_path_for("update_project", project_id=default_project_id),
            headers=headers,
            json={"is_public": False},
        )
        assert response.status_code == 422

    async def test_updates_timestamps(
        self, client: AsyncClient, default_public_project: Project, make_org_user, session: Session
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        project = session.get(Project, default_project_id)
        created_at = project.created_at

        response = await client.post(
            app.url_path_for("update_project", project_id=default_project_id),
            headers=headers,
            json={"name": "Coucou 2", "is_public": False},
        )
        assert response.status_code == 200
        session.refresh(project)
        assert project.name == "Coucou 2"
        assert project.created_at == created_at
        assert project.created_at < project.updated_at
        assert not project.is_public


class TestDeleteProject:
    @pytest.mark.parametrize(
        "role, expected_status",
        [
            (OrgRole.ORG_ADMIN, 200),
            (OrgRole.RESPO_BENEVOLE, 403),
            (OrgRole.RESPO_COMMISSION, 403),
            (None, 403),
        ],
    )
    async def test_role_access(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
        session: Session,
        role,
        expected_status,
    ):
        _, headers = make_org_user(role=role)
        response = await client.delete(
            app.url_path_for("delete_project", project_id=default_project_id), headers=headers
        )
        assert response.status_code == expected_status
        project = session.get(Project, default_project_id)
        if expected_status == 200:
            assert project is None
        else:
            assert project is not None

    async def test_cascades_to_volunteers_and_slots(
        self, client: AsyncClient, default_public_project: Project, make_org_user, session: Session
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        response = await client.delete(
            app.url_path_for("delete_project", project_id=default_project_id), headers=headers
        )
        assert response.status_code == 200
        assert (
            session.execute(select(Volunteer).where(Volunteer.project_id == default_project_id))
            .scalars()
            .first()
            is None
        )
        assert (
            session.execute(select(Slot).where(Slot.project_id == default_project_id))
            .scalars()
            .first()
            is None
        )