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

from uuid import uuid4

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

from app.main import app
from app.models import Commission, CommissionMember, OrgRole, Project, Slot, SlotTemplate
from app.tests.conftest import default_slot_id, default_tag_id, default_template_id

pytestmark = pytest.mark.asyncio

WRITE_FORBIDDEN_ROLES = [OrgRole.RESPO_COMMISSION, None]  # respo_commission with NO commission set


@pytest.fixture
def commission_and_member(session: Session, default_public_project: Project, make_org_user):
    """A commission in default_public_project, with a fresh respo_commission
    user as its member. Returns (commission, member_user, member_headers)."""
    commission = Commission(project_id=default_public_project.id, name="Bar")
    session.add(commission)
    session.commit()
    user, headers = make_org_user(role=OrgRole.RESPO_COMMISSION)
    session.add(CommissionMember(commission_id=commission.id, user_id=user.id))
    session.commit()
    return commission, user, headers


@pytest.fixture
def other_commission(session: Session, default_public_project: Project):
    """A second commission the respo_commission fixture user is NOT part of."""
    commission = Commission(project_id=default_public_project.id, name="Scene")
    session.add(commission)
    session.commit()
    return commission


class TestCreateTemplate:
    async def test_requires_auth(self, client: AsyncClient, default_public_project: Project):
        response = await client.post(
            app.url_path_for("create_template", project_id=default_public_project.id),
            json={"title": "1st template"},
        )
        assert response.status_code == 401

    async def test_other_org_member_forbidden(
        self, client: AsyncClient, default_public_project: Project, other_org_user
    ):
        _, headers = other_org_user
        response = await client.post(
            app.url_path_for("create_template", project_id=default_public_project.id),
            json={"title": "1st template"},
            headers=headers,
        )
        assert response.status_code == 403

    async def test_validation_error(
        self,
        client: AsyncClient,
        default_public_project: Project,
        default_user_headers: dict,
        default_user_org_membership,
    ):
        response = await client.post(
            app.url_path_for("create_template", project_id=default_public_project.id),
            json={},
            headers=default_user_headers,
        )
        assert response.status_code == 422

    async def test_project_not_found(
        self,
        client: AsyncClient,
        default_user_headers: dict,
        default_user_org_membership,
    ):
        response = await client.post(
            app.url_path_for("create_template", project_id=uuid4()),
            json={"title": "1st template"},
            headers=default_user_headers,
        )
        assert response.status_code == 404

    @pytest.mark.parametrize(
        "payload",
        [
            {"title": "1st template"},
            {"title": "1𝖘ҭ ṥ٥ι𝙪𝓉ìóη", "place": "echo"},
            {"title": "1st template", "responsible_override": "echo"},
            {"title": "1st template", "description": "&é'(-è_çecho"},
            {"title": "1st template", "place": "3", "description": "&é'(-è_çecho"},
            {"title": "1st template", "comment": "55"},
        ],
    )
    async def test_org_admin_creates(
        self,
        client: AsyncClient,
        default_public_project: Project,
        default_user_headers: dict,
        default_user_org_membership,
        session: Session,
        payload: dict,
    ):
        response = await client.post(
            app.url_path_for("create_template", project_id=default_public_project.id),
            json=payload,
            headers=default_user_headers,
        )
        assert response.status_code == 200
        template = session.get(SlotTemplate, response.json()["id"])
        assert template is not None
        for k in ["title", "description", "place"]:
            val = getattr(template, k)
            assert val == payload[k] if k in payload else val == ""

    async def test_respo_benevole_creates_without_commission(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
    ):
        _, headers = make_org_user(role=OrgRole.RESPO_BENEVOLE)
        response = await client.post(
            app.url_path_for("create_template", project_id=default_public_project.id),
            json={"title": "Benevole template"},
            headers=headers,
        )
        assert response.status_code == 200

    async def test_respo_commission_creates_for_own_commission(
        self,
        client: AsyncClient,
        default_public_project: Project,
        commission_and_member,
    ):
        commission, _, headers = commission_and_member
        response = await client.post(
            app.url_path_for("create_template", project_id=default_public_project.id),
            json={"title": "Bar setup", "commission_id": commission.id},
            headers=headers,
        )
        assert response.status_code == 200
        assert response.json()["commission_id"] == commission.id

    async def test_respo_commission_rejected_for_other_commission(
        self,
        client: AsyncClient,
        default_public_project: Project,
        commission_and_member,
        other_commission,
    ):
        _, _, headers = commission_and_member
        response = await client.post(
            app.url_path_for("create_template", project_id=default_public_project.id),
            json={"title": "Scene setup", "commission_id": other_commission.id},
            headers=headers,
        )
        assert response.status_code == 403

    async def test_respo_ok(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
    ):
        _, headers = make_org_user(role=OrgRole.RESPO_COMMISSION)
        response = await client.post(
            app.url_path_for("create_template", project_id=default_public_project.id),
            json={"title": "No commission"},
            headers=headers,
        )
        assert response.status_code == 200


class TestUpdateTemplate:
    async def test_requires_auth(self, client: AsyncClient, default_public_project: Project):
        response = await client.post(
            app.url_path_for(
                "update_template",
                project_id=default_public_project.id,
                template_id=default_template_id,
            ),
            json={"title": "x"},
        )
        assert response.status_code == 401

    async def test_other_org_member_forbidden(
        self, client: AsyncClient, default_public_project: Project, other_org_user
    ):
        _, headers = other_org_user
        response = await client.post(
            app.url_path_for(
                "update_template",
                project_id=default_public_project.id,
                template_id=default_template_id,
            ),
            json={"title": "x"},
            headers=headers,
        )
        assert response.status_code == 403

    async def test_invalid_template_id_format(
        self,
        client: AsyncClient,
        default_public_project: Project,
        default_user_headers: dict,
        default_user_org_membership,
    ):
        response = await client.post(
            app.url_path_for(
                "update_template", project_id=default_public_project.id, template_id="12"
            ),
            json={"title": "1st template"},
            headers=default_user_headers,
        )
        assert response.status_code == 422

    async def test_template_not_found(
        self,
        client: AsyncClient,
        default_public_project: Project,
        default_user_headers: dict,
        default_user_org_membership,
    ):
        response = await client.post(
            app.url_path_for(
                "update_template", project_id=default_public_project.id, template_id=uuid4()
            ),
            json={"title": "1st template"},
            headers=default_user_headers,
        )
        assert response.status_code == 404

    async def test_project_not_found(
        self,
        client: AsyncClient,
        default_user_headers: dict,
        default_user_org_membership,
    ):
        response = await client.post(
            app.url_path_for(
                "update_template", project_id=uuid4(), template_id=default_template_id
            ),
            json={"title": "1st template"},
            headers=default_user_headers,
        )
        assert response.status_code == 404

    @pytest.mark.parametrize(
        "code,payload",
        [
            (422, {"title": [1.001, 2]}),
            (422, {"responsible_override": {"t": None}}),
            (422, {"place": [1, 2]}),
            (422, {"tags": ["1", "2"]}),
            (400, {"tags": [default_slot_id]}),
            (422, {"comment": 1235}),
        ],
    )
    async def test_payload_validation(
        self,
        client: AsyncClient,
        default_public_project: Project,
        default_user_headers: dict,
        default_user_org_membership,
        payload: dict,
        code: int,
    ):
        response = await client.post(
            app.url_path_for(
                "update_template",
                project_id=default_public_project.id,
                template_id=default_template_id,
            ),
            json=payload,
            headers=default_user_headers,
        )
        assert response.status_code == code

    @pytest.mark.parametrize(
        "payload",
        [
            {"title": "1st template"},
            {"title": "1st template", "place": "echo"},
            {"title": "1st template", "responsible_override": "echo"},
            {"title": "1st template", "description": "&é'(-è_çecho"},
            {"title": "1st template", "place": "Ḽơᶉëᶆ ȋṕšᶙṁ", "description": "&é'(-è_çecho"},
            {"title": "1st template", "place": "3", "description": "&é'(-è_çecho"},
            {"title": "1st template", "comment": "&é'(-è_çecho"},
        ],
    )
    async def test_org_admin_updates(
        self,
        client: AsyncClient,
        default_public_project: Project,
        default_user_headers: dict,
        default_user_org_membership,
        session: Session,
        payload: dict,
    ):
        response = await client.post(
            app.url_path_for(
                "update_template",
                project_id=default_public_project.id,
                template_id=default_template_id,
            ),
            json=payload,
            headers=default_user_headers,
        )
        assert response.status_code == 200
        template = session.get(SlotTemplate, default_template_id)
        assert template is not None
        for k in ["title", "description", "place", "comment"]:
            val = getattr(template, k)
            assert val == payload[k] if k in payload else val == ""

    async def test_update_tags(
        self,
        client: AsyncClient,
        default_public_project: Project,
        default_user_headers: dict,
        default_user_org_membership,
        session: Session,
    ):
        url = app.url_path_for(
            "update_template", project_id=default_public_project.id, template_id=default_template_id
        )
        response = await client.post(
            url, json={"tags": [default_tag_id]}, headers=default_user_headers
        )
        assert response.status_code == 200
        template = session.get(SlotTemplate, default_template_id)
        assert len(template.tags) == 1
        assert template.tags[0].id == default_tag_id

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

    async def test_respo_commission_updates_own_template(
        self,
        client: AsyncClient,
        default_public_project: Project,
        commission_and_member,
        session: Session,
    ):
        commission, _, headers = commission_and_member
        template = SlotTemplate(
            project_id=default_public_project.id, title="Bar", commission_id=commission.id
        )
        session.add(template)
        session.commit()

        response = await client.post(
            app.url_path_for(
                "update_template", project_id=default_public_project.id, template_id=template.id
            ),
            json={"title": "Bar renamed"},
            headers=headers,
        )
        assert response.status_code == 200
        session.refresh(template)
        assert template.title == "Bar renamed"

    async def test_respo_commission_rejected_for_other_commission_template(
        self,
        client: AsyncClient,
        default_public_project: Project,
        commission_and_member,
        other_commission,
        session: Session,
    ):
        _, _, headers = commission_and_member
        template = SlotTemplate(
            project_id=default_public_project.id, title="Scene", commission_id=other_commission.id
        )
        session.add(template)
        session.commit()

        response = await client.post(
            app.url_path_for(
                "update_template", project_id=default_public_project.id, template_id=template.id
            ),
            json={"title": "Hijacked"},
            headers=headers,
        )
        assert response.status_code == 403

    async def test_respo_commission_reassign_to_other_commission_rejected(
        self,
        client: AsyncClient,
        default_public_project: Project,
        commission_and_member,
        other_commission,
        session: Session,
    ):
        commission, _, headers = commission_and_member
        template = SlotTemplate(
            project_id=default_public_project.id, title="Bar", commission_id=commission.id
        )
        session.add(template)
        session.commit()

        response = await client.post(
            app.url_path_for(
                "update_template", project_id=default_public_project.id, template_id=template.id
            ),
            json={"commission_id": other_commission.id},
            headers=headers,
        )
        assert response.status_code == 403


class TestDeleteTemplate:
    async def test_requires_auth(self, client: AsyncClient, default_public_project: Project):
        response = await client.delete(
            app.url_path_for(
                "delete_template",
                project_id=default_public_project.id,
                template_id=default_template_id,
            )
        )
        assert response.status_code == 401

    async def test_other_org_member_forbidden(
        self, client: AsyncClient, default_public_project: Project, other_org_user
    ):
        response = await client.delete(
            app.url_path_for(
                "delete_template",
                project_id=default_public_project.id,
                template_id=default_template_id,
            ),
            headers=other_org_user[1],
        )
        assert response.status_code == 403

    async def test_invalid_ids(
        self,
        client: AsyncClient,
        default_public_project: Project,
        default_user_headers: dict,
        default_user_org_membership,
    ):
        response = await client.delete(
            app.url_path_for(
                "delete_template",
                project_id=default_public_project.id,
                template_id="default_tag_id",
            ),
            headers=default_user_headers,
        )
        assert response.status_code == 422
        response = await client.delete(
            app.url_path_for("delete_template", project_id="ded", template_id=default_template_id),
            headers=default_user_headers,
        )
        assert response.status_code == 422

    async def test_org_admin_deletes(
        self,
        client: AsyncClient,
        default_public_project: Project,
        default_user_headers: dict,
        default_user_org_membership,
        session: Session,
    ):
        response = await client.delete(
            app.url_path_for(
                "delete_template",
                project_id=default_public_project.id,
                template_id=default_template_id,
            ),
            headers=default_user_headers,
        )
        assert response.status_code == 200
        assert session.get(SlotTemplate, default_template_id) is None

    async def test_delete_does_not_delete_slot(
        self,
        client: AsyncClient,
        default_public_project: Project,
        default_user_headers: dict,
        default_user_org_membership,
        session: Session,
    ):
        slot = session.get(Slot, default_slot_id)
        slot.template_id = default_template_id
        session.commit()

        response = await client.delete(
            app.url_path_for(
                "delete_template",
                project_id=default_public_project.id,
                template_id=default_template_id,
            ),
            headers=default_user_headers,
        )
        assert response.status_code == 200
        assert session.get(Slot, default_slot_id) is not None

    async def test_respo_commission_deletes_own_template(
        self,
        client: AsyncClient,
        default_public_project: Project,
        commission_and_member,
        session: Session,
    ):
        commission, _, headers = commission_and_member
        template = SlotTemplate(
            project_id=default_public_project.id, title="Bar", commission_id=commission.id
        )
        session.add(template)
        session.commit()
        template_id = template.id
        session.expunge(template)

        response = await client.delete(
            app.url_path_for(
                "delete_template", project_id=default_public_project.id, template_id=template_id
            ),
            headers=headers,
        )
        assert response.status_code == 200
        assert session.get(SlotTemplate, template_id) is None

    async def test_respo_commission_rejected_for_other_commission(
        self,
        client: AsyncClient,
        default_public_project: Project,
        commission_and_member,
        other_commission,
        session: Session,
    ):
        _, _, headers = commission_and_member
        template = SlotTemplate(
            project_id=default_public_project.id, title="Scene", commission_id=other_commission.id
        )
        session.add(template)
        session.commit()

        response = await client.delete(
            app.url_path_for(
                "delete_template", project_id=default_public_project.id, template_id=template.id
            ),
            headers=headers,
        )
        assert response.status_code == 403
        assert session.get(SlotTemplate, template.id) is not None