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

import uuid
from datetime import datetime

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, Project, Slot, SlotTag, SlotTemplate
from app.tests.conftest import default_organization_id, default_project_id, default_tag_id

pytestmark = pytest.mark.asyncio

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


class TestListProjectTags:
    async def test_requires_auth(self, client: AsyncClient, default_public_project: Project):
        response = await client.get(
            app.url_path_for("list_project_tags", project_id=default_project_id)
        )
        assert response.status_code == 401

    @pytest.mark.parametrize("role", ALL_ROLES)
    async def test_all_roles_can_read(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
        session: Session,
        role,
    ):
        _, headers = make_org_user(role=role)
        response = await client.get(
            app.url_path_for("list_project_tags", project_id=default_project_id), headers=headers
        )
        assert response.status_code == 200
        assert len(response.json()) == 1

        tag = SlotTag(title="1er tag", project_id=default_project_id)
        session.add(tag)
        session.commit()

        response = await client.get(
            app.url_path_for("list_project_tags", project_id=default_project_id), headers=headers
        )
        assert response.status_code == 200
        assert len(response.json()) == 2

    async def test_no_membership_forbidden(
        self, client: AsyncClient, default_public_project: Project, make_org_user
    ):
        _, headers = make_org_user(role=None)
        response = await client.get(
            app.url_path_for("list_project_tags", project_id=default_project_id), headers=headers
        )
        assert response.status_code == 403

    async def test_other_org_member_forbidden(
        self, client: AsyncClient, default_public_project: Project, other_org_user
    ):
        _, headers = other_org_user
        response = await client.get(
            app.url_path_for("list_project_tags", project_id=default_project_id), headers=headers
        )
        assert response.status_code == 403


class TestCreateTag:
    async def test_requires_auth(self, client: AsyncClient, default_public_project: Project):
        response = await client.post(app.url_path_for("create_tag", project_id=default_project_id))
        assert response.status_code == 401

    async def test_invalid_project_id_format(
        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("create_tag", project_id="default_project_id"), headers=headers
        )
        assert response.status_code == 422

    async def test_project_not_found(
        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("create_tag", project_id=uuid.uuid4()),
            json={"title": "1st tag"},
            headers=headers,
        )
        assert response.status_code == 404

    async def test_invalid_payload(
        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("create_tag", project_id=default_project_id),
            json={"titl": "1st tag"},
            headers=headers,
        )
        assert response.status_code == 422

    @pytest.mark.parametrize("role", ALL_ROLES)
    async def test_all_roles_can_create(
        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_tag", project_id=default_project_id),
            json={"title": "1st tag"},
            headers=headers,
        )
        assert response.status_code == 200
        tag = (
            session.execute(select(SlotTag).where(SlotTag.project_id == default_project_id))
            .scalars()
            .first()
        )
        assert tag is not None
        assert tag.title == "1st tag"

        response = await client.post(
            app.url_path_for("create_tag", project_id=default_project_id),
            json={"title": "1st tag"},
            headers=headers,
        )
        assert response.status_code == 200
        tag_count = len(
            session.execute(select(SlotTag).where(SlotTag.project_id == default_project_id))
            .scalars()
            .all()
        )
        assert tag_count == 2

    async def test_no_membership_forbidden(
        self, client: AsyncClient, default_project: Project, make_org_user
    ):
        _, headers = make_org_user(role=None)
        response = await client.post(
            app.url_path_for("create_tag", project_id=default_project_id),
            json={"title": "1st tag"},
            headers=headers,
        )
        assert response.status_code == 403

    async def test_create_with_template(
        self, client: AsyncClient, default_public_project: Project, make_org_user, session: Session
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        template = SlotTemplate(
            project_id=default_project_id, title="coucou", description="ceci est une description"
        )
        session.add(template)
        session.commit()

        response = await client.post(
            app.url_path_for("create_tag", project_id=default_project_id),
            json={"title": "1st tag", "templates": [template.id]},
            headers=headers,
        )
        assert response.status_code == 200
        assert response.json()["templates_id"][0] == template.id
        session.refresh(template)
        assert len(template.tags) > 0

    async def test_create_with_template_from_other_project_rejected(
        self, client: AsyncClient, default_public_project: Project, make_org_user, session: Session
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        template = SlotTemplate(
            project_id=default_project_id, title="coucou", description="ceci est une description"
        )
        session.add(template)
        session.commit()

        other_project = Project(
            organization_id=default_organization_id, name="second project", is_public=False
        )
        session.add(other_project)
        session.commit()

        response = await client.post(
            app.url_path_for("create_tag", project_id=other_project.id),
            json={"title": "1st tag", "templates": [template.id]},
            headers=headers,
        )
        assert response.status_code == 400

    async def test_other_org_member_forbidden(
        self, client: AsyncClient, default_project: Project, other_org_user
    ):
        _, headers = other_org_user
        response = await client.post(
            app.url_path_for("create_tag", project_id=default_project_id),
            json={"title": "hijacked"},
            headers=headers,
        )
        assert response.status_code == 403


class TestUpdateTag:
    async def test_requires_auth(self, client: AsyncClient, default_public_project: Project):
        url = app.url_path_for("update_tag", project_id=default_project_id, tag_id=default_tag_id)
        response = await client.post(url, json={"title": "royaux"})
        assert response.status_code == 401

    @pytest.mark.parametrize(
        "payload", [{"title": [1.001, 2]}, {"title": {}}, {"templates": ["1", "2"]}]
    )
    async def test_invalid_payload(
        self, client: AsyncClient, default_public_project: Project, make_org_user, payload: dict
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        url = app.url_path_for("update_tag", project_id=default_project_id, tag_id=default_tag_id)
        response = await client.post(url, json=payload, headers=headers)
        assert response.status_code == 422

    @pytest.mark.parametrize("role", ALL_ROLES)
    async def test_all_roles_can_update(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
        session: Session,
        role,
    ):
        _, headers = make_org_user(role=role)
        tag = session.get(SlotTag, default_tag_id)
        assert tag is not None

        url = app.url_path_for("update_tag", project_id=default_project_id, tag_id=default_tag_id)
        response = await client.post(url, json={"title": "new_title"}, headers=headers)
        assert response.status_code == 200
        session.refresh(tag)
        assert tag.title == "new_title"
        assert response.json()["title"] == "new_title"

        template = SlotTemplate(project_id=default_project_id, title="template")
        session.add(template)
        session.commit()

        response = await client.post(url, json={"templates": [template.id]}, headers=headers)
        assert response.status_code == 200
        session.refresh(tag)
        assert len(tag.templates) == 1

        response = await client.post(url, json={"templates": []}, headers=headers)
        assert response.status_code == 200
        session.refresh(tag)
        assert len(tag.templates) == 0

    async def test_no_membership_forbidden(
        self, client: AsyncClient, default_public_project: Project, make_org_user, session: Session
    ):
        _, headers = make_org_user(role=None)
        tag = session.get(SlotTag, default_tag_id)
        url = app.url_path_for("update_tag", project_id=default_project_id, tag_id=default_tag_id)
        response = await client.post(url, json={"title": "hijacked"}, headers=headers)
        assert response.status_code == 403
        session.refresh(tag)
        assert tag.title != "hijacked"

    async def test_template_from_other_project_rejected(
        self, client: AsyncClient, default_public_project: Project, make_org_user, session: Session
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        tag = session.get(SlotTag, default_tag_id)
        assert tag is not None

        other_project = Project(
            organization_id=default_organization_id, name="other project", is_public=False
        )
        template = SlotTemplate(title="template 1")
        other_project.templates.append(template)
        session.add(template)
        session.add(other_project)
        session.commit()

        url = app.url_path_for("update_tag", project_id=default_project_id, tag_id=default_tag_id)
        response = await client.post(url, json={"templates": [template.id]}, headers=headers)
        assert response.status_code == 400

    async def test_other_org_member_forbidden(
        self, client: AsyncClient, default_public_project: Project, other_org_user, session: Session
    ):
        _, headers = other_org_user
        tag = session.get(SlotTag, default_tag_id)
        url = app.url_path_for("update_tag", project_id=default_project_id, tag_id=default_tag_id)
        response = await client.post(url, json={"title": "hijacked"}, headers=headers)
        assert response.status_code == 403
        session.refresh(tag)
        assert tag.title != "hijacked"


class TestListTaggedSlot:
    @pytest.mark.parametrize("n_slot", [1, 5, 10])
    async def test_returns_slots_for_tag(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
        session: Session,
        n_slot: int,
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        url = app.url_path_for(
            "list_tagged_slot", project_id=default_project_id, tag_id=default_tag_id
        )

        response = await client.get(url, headers=headers)
        assert response.status_code == 200
        assert len(response.json()) == 0

        tag = session.get(SlotTag, default_tag_id)
        assert tag is not None
        for i in range(n_slot):
            template = SlotTemplate(project_id=default_project_id, title=f"template {i}")
            slot = Slot(
                project_id=default_project_id,
                title=f"Slot {i}",
                starting_time=datetime(2024, 9, 9, 12, 2 * i),
                ending_time=datetime(2024, 9, 9, 12, 2 * (i + 1)),
            )
            slot.template = template
            tag.templates.append(template)
            session.add_all([template, slot])
        session.commit()

        response = await client.get(url, headers=headers)
        assert response.status_code == 200
        assert len(response.json()) == n_slot

    async def test_no_membership_forbidden(
        self, client: AsyncClient, default_public_project: Project, make_org_user
    ):
        _, headers = make_org_user(role=None)
        url = app.url_path_for(
            "list_tagged_slot", project_id=default_project_id, tag_id=default_tag_id
        )
        response = await client.get(url, headers=headers)
        assert response.status_code == 403

    async def test_other_org_member_forbidden(
        self, client: AsyncClient, default_public_project: Project, other_org_user
    ):
        _, headers = other_org_user
        url = app.url_path_for(
            "list_tagged_slot", project_id=default_project_id, tag_id=default_tag_id
        )
        response = await client.get(url, headers=headers)
        assert response.status_code == 403


class TestDeleteTag:
    async def test_requires_auth(self, client: AsyncClient, default_public_project: Project):
        url = app.url_path_for("delete_tag", project_id=default_project_id, tag_id=default_tag_id)
        response = await client.delete(url)
        assert response.status_code == 401

    async def test_invalid_tag_id_format(
        self, client: AsyncClient, default_public_project: Project, make_org_user
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        url = app.url_path_for("delete_tag", project_id=default_project_id, tag_id="default_tag_id")
        response = await client.delete(url, headers=headers)
        assert response.status_code == 422

    async def test_invalid_project_id_format(
        self, client: AsyncClient, default_public_project: Project, make_org_user
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        url = app.url_path_for(
            "delete_tag", project_id="default_volunteer_id", tag_id=default_tag_id
        )
        response = await client.delete(url, headers=headers)
        assert response.status_code == 422

    async def test_no_membership_forbidden(
        self, client: AsyncClient, default_public_project: Project, make_org_user, session: Session
    ):
        _, headers = make_org_user(role=None)
        url = app.url_path_for("delete_tag", project_id=default_project_id, tag_id=default_tag_id)
        response = await client.delete(url, headers=headers)
        assert response.status_code == 403
        assert session.get(SlotTag, default_tag_id) is not None

    @pytest.mark.parametrize("role", ALL_ROLES)
    async def test_all_roles_can_delete(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
        session: Session,
        role,
    ):
        _, headers = make_org_user(role=role)
        url = app.url_path_for("delete_tag", project_id=default_project_id, tag_id=default_tag_id)
        response = await client.delete(url, headers=headers)
        assert response.status_code == 200
        assert session.get(SlotTag, default_tag_id) is None

    async def test_idempotent_delete(
        self, client: AsyncClient, default_public_project: Project, make_org_user
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        url = app.url_path_for("delete_tag", project_id=default_project_id, tag_id=default_tag_id)
        response = await client.delete(url, headers=headers)
        assert response.status_code == 200
        response = await client.delete(url, headers=headers)
        assert response.status_code == 200

    async def test_delete_random_uuid(
        self, client: AsyncClient, default_public_project: Project, make_org_user
    ):
        """tags.py still uses bulk DELETE with no existence check -- same
        idempotent-200-on-missing pattern as volunteers.py/sms.py, not the
        404-via-dependency-resolution used by slots.py/templates.py."""
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        url = app.url_path_for("delete_tag", project_id=default_project_id, tag_id=uuid.uuid4())
        response = await client.delete(url, headers=headers)
        assert response.status_code == 200

    async def test_other_org_member_forbidden(
        self, client: AsyncClient, default_public_project: Project, other_org_user, session: Session
    ):
        _, headers = other_org_user
        url = app.url_path_for("delete_tag", project_id=default_project_id, tag_id=default_tag_id)
        response = await client.delete(url, headers=headers)
        assert response.status_code == 403
        assert session.get(SlotTag, default_tag_id) is not None