bdlg2023-sms-back/app/tests/test_templates.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, OrgRole, Project, Slot, SlotTemplate
from app.tests.conftest import (
    default_project_id,
    default_slot_id,
    default_tag_id,
    default_template_id,
)
from app.tests.shared_access import SharedProjectAccessTests

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)
    commission.members.append(user)
    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


route_kwargs = {"project_id": default_project_id}
route_kwargs_2 = {**route_kwargs, "template_id": default_template_id}
TEMPLATE_ROUTES = [
    ("GET", "list_project_templates", route_kwargs, None),
    ("POST", "create_template", route_kwargs, {"title": "Test"}),
    ("POST", "update_template", route_kwargs_2, {"title": "Updated"}),
    ("DELETE", "delete_template", route_kwargs_2, None),
]


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


class TestListProjectTemplate:
    async def test_ok(self, client: AsyncClient, default_public_project: Project, make_org_user):
        _, headers = make_org_user(OrgRole.ORG_ADMIN)
        response = await client.get(
            app.url_path_for("list_project_templates", project_id=default_public_project.id),
            headers=headers,
        )
        assert response.status_code == 200
        assert len(response.json()) > 0


class TestCreateTemplate:
    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

    @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_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

    @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_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