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

from datetime import UTC, datetime, timedelta
from uuid import uuid4

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

from app.main import app
from app.models import (
    Commission,
    OrgRole,
    Project,
    Slot,
    SlotTemplate,
    Volunteer,
)
from app.tests.conftest import (
    default_project_id,
    default_slot_id,
    default_template_id,
    default_volunteer_id,
)
from app.tests.shared_access import SharedProjectAccessTests

pytestmark = pytest.mark.asyncio


@pytest.fixture
def commission_and_member(session: Session, default_public_project: Project, make_org_user):
    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):
    commission = Commission(project_id=default_public_project.id, name="Scene")
    session.add(commission)
    session.commit()
    return commission


@pytest.fixture
def commission_template(
    session: Session, default_public_project: Project, commission_and_member
) -> SlotTemplate:
    commission, _, _ = commission_and_member
    template = SlotTemplate(
        project_id=default_public_project.id, title="Bar template", commission_id=commission.id
    )
    session.add(template)
    session.commit()
    return template


@pytest.fixture
def other_commission_template(
    session: Session, default_public_project: Project, other_commission
) -> SlotTemplate:
    template = SlotTemplate(
        project_id=default_public_project.id,
        title="Scene template",
        commission_id=other_commission.id,
    )
    session.add(template)
    session.commit()
    return template


t1 = datetime(2000, 1, 1, tzinfo=UTC)
SLOT_ROUTES = [
    ("GET", "list_project_slots", {"project_id": "PROJECT"}, None),
    (
        "POST",
        "create_slot",
        {"project_id": "PROJECT"},
        {
            "title": "Test",
            "starting_time": t1.isoformat(),
            "ending_time": (t1 + timedelta(days=1)).isoformat(),
        },
    ),
    (
        "POST",
        "update_slot",
        {"project_id": "PROJECT", "slot_id": default_slot_id},
        {"title": "Updated"},
    ),
    ("DELETE", "delete_slot", {"project_id": "PROJECT", "slot_id": default_slot_id}, None),
]


class TestSlotCrossCuttingAccess(SharedProjectAccessTests):
    @pytest.fixture(params=SLOT_ROUTES, ids=lambda x: f"{x[0]}-{x[1]}")
    def resolved_route(self, request, default_project):
        method, endpoint, kwargs, payload = request.param
        resolved_kwargs = {
            k: (default_project.id if v == "PROJECT" else v) for k, v in kwargs.items()
        }
        return method, endpoint, resolved_kwargs, payload


class TestListProjectSlots:
    async def test_invalid_project_id(
        self, client: AsyncClient, default_user_headers: dict, default_user_org_membership
    ):
        response = await client.get(
            app.url_path_for("list_project_slots", project_id="pas un uuid valid"),
            headers=default_user_headers,
        )
        assert response.status_code == 422

    async def test_returns_slots(
        self,
        client: AsyncClient,
        default_public_project: Project,
        default_user_headers: dict,
        default_user_org_membership,
    ):
        response = await client.get(
            app.url_path_for("list_project_slots", project_id=default_project_id),
            headers=default_user_headers,
        )
        assert response.status_code == 200
        data = response.json()
        assert len(data) == 1
        assert data[0]["title"] == "être roi"
        assert data[0]["id"] == default_slot_id
        assert "created_at" in data[0]

    async def test_respo_commission_sees_all_project_slots(
        self,
        client: AsyncClient,
        default_public_project: Project,
        commission_and_member,
    ):
        """respo_commission has read access to the whole plan, not just
        their own commission's slots."""
        _, _, headers = commission_and_member
        response = await client.get(
            app.url_path_for("list_project_slots", project_id=default_project_id), headers=headers
        )
        assert response.status_code == 200
        assert len(response.json()) == 1

    async def test_cannot_look_other_org_slot(
        self,
        client: AsyncClient,
        default_public_project: Project,
        session: Session,
        commission_and_member,
        other_org_user,
    ):
        """An admin of a project belonging to a DIFFERENT organization cannot access the project."""
        _, headers = other_org_user
        response = await client.get(
            app.url_path_for("list_project_slots", project_id=default_public_project.id),
            headers=headers,
        )
        assert response.status_code == 403


class TestCreateSlot:
    async def test_org_admin_creates(
        self,
        client: AsyncClient,
        default_public_project: Project,
        default_user_headers: dict,
        default_user_org_membership,
        session: Session,
    ):
        starting_time = datetime(1900, 1, 1)
        payload = {
            "title": "être mort",
            "starting_time": starting_time.isoformat(),
            "ending_time": (starting_time + timedelta(minutes=60)).isoformat(),
        }
        response = await client.post(
            app.url_path_for("create_slot", project_id=default_project_id),
            json=payload,
            headers=default_user_headers,
        )
        assert response.status_code == 200
        assert response.json()["id"] != default_slot_id
        assert response.json()["title"] == "être mort"
        assert response.json()["required_volunteers"] == 0

        result = session.execute(select(Slot).where(Slot.project_id == default_project_id))
        slots = result.scalars().all()
        assert len(slots) > 1
        slot = [s for s in slots if s.id != default_slot_id][0]
        assert slot.title == "être mort"
        assert abs(slot.starting_time - starting_time.replace(tzinfo=UTC)) < timedelta(minutes=30)

        del payload["title"]
        response = await client.post(
            app.url_path_for("create_slot", project_id=default_project_id),
            json=payload,
            headers=default_user_headers,
        )
        assert response.status_code == 422

    async def test_min_volunteer(
        self,
        client: AsyncClient,
        default_public_project: Project,
        default_user_headers: dict,
        default_user_org_membership,
        session: Session,
    ):
        starting_time = datetime(1900, 1, 1)
        payload = {
            "title": "être mort",
            "starting_time": starting_time.isoformat(),
            "ending_time": (starting_time + timedelta(minutes=60)).isoformat(),
            "required_volunteers": 2,
        }
        response = await client.post(
            app.url_path_for("create_slot", project_id=default_project_id),
            json=payload,
            headers=default_user_headers,
        )
        assert response.status_code == 200
        slots = (
            session.execute(select(Slot).where(Slot.project_id == default_project_id))
            .scalars()
            .all()
        )
        slot = [s for s in slots if s.id != default_slot_id][0]
        assert slot.required_volunteers == 2

    async def test_respo_benevole_creates_without_template(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
    ):
        _, headers = make_org_user(role=OrgRole.RESPO_BENEVOLE)
        starting_time = datetime(1900, 1, 1)
        response = await client.post(
            app.url_path_for("create_slot", project_id=default_project_id),
            json={
                "title": "Benevole slot",
                "starting_time": starting_time.isoformat(),
                "ending_time": (starting_time + timedelta(minutes=60)).isoformat(),
            },
            headers=headers,
        )
        assert response.status_code == 200

    async def test_respo_commission_creates_with_own_template(
        self,
        client: AsyncClient,
        default_public_project: Project,
        commission_and_member,
        commission_template,
    ):
        _, _, headers = commission_and_member
        starting_time = datetime(1900, 1, 1)
        response = await client.post(
            app.url_path_for("create_slot", project_id=default_project_id),
            json={
                "title": "Bar shift",
                "starting_time": starting_time.isoformat(),
                "ending_time": (starting_time + timedelta(minutes=60)).isoformat(),
                "template_id": commission_template.id,
            },
            headers=headers,
        )
        assert response.status_code == 200
        assert response.json()["template_id"] == commission_template.id

    async def test_respo_commission_rejected_with_other_commission_template(
        self,
        client: AsyncClient,
        default_public_project: Project,
        commission_and_member,
        other_commission_template,
    ):
        _, _, headers = commission_and_member
        starting_time = datetime(1900, 1, 1)
        response = await client.post(
            app.url_path_for("create_slot", project_id=default_project_id),
            json={
                "title": "Scene shift",
                "starting_time": starting_time.isoformat(),
                "ending_time": (starting_time + timedelta(minutes=60)).isoformat(),
                "template_id": other_commission_template.id,
            },
            headers=headers,
        )
        assert response.status_code == 403

    async def test_with_invalid_volunteer(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
    ):
        _, headers = make_org_user(role=OrgRole.RESPO_BENEVOLE)
        starting_time = datetime(1900, 1, 1)
        response = await client.post(
            app.url_path_for("create_slot", project_id=default_project_id),
            json={
                "title": "Scene shift",
                "starting_time": starting_time.isoformat(),
                "ending_time": (starting_time + timedelta(minutes=60)).isoformat(),
                "volunteers": [str(uuid4())],
            },
            headers=headers,
        )
        assert response.status_code == 400

    async def test_with_volunteer(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
    ):
        _, headers = make_org_user(role=OrgRole.RESPO_BENEVOLE)
        starting_time = datetime(1900, 1, 1)
        response = await client.post(
            app.url_path_for("create_slot", project_id=default_project_id),
            json={
                "title": "Scene shift",
                "starting_time": starting_time.isoformat(),
                "ending_time": (starting_time + timedelta(minutes=60)).isoformat(),
                "volunteers": [default_volunteer_id],
            },
            headers=headers,
        )
        assert response.status_code == 200


class TestUpdateSlot:
    async def test_slot_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_slot", project_id=default_project_id, slot_id=uuid4()),
            json={"title": "x"},
            headers=default_user_headers,
        )
        assert response.status_code == 404

    async def test_org_admin_updates_each_field(
        self,
        client: AsyncClient,
        default_public_project: Project,
        default_user_headers: dict,
        default_user_org_membership,
    ):
        starting_time = datetime(2000, 1, 1, tzinfo=UTC)
        payload = {
            "title": "être mort 2 fois",
            "starting_time": starting_time.isoformat(),
            "ending_time": (starting_time + timedelta(minutes=60)).isoformat(),
            "required_volunteers": 2,
        }
        for k, v in payload.items():
            response = await client.post(
                app.url_path_for(
                    "update_slot", project_id=default_project_id, slot_id=default_slot_id
                ),
                json={k: v},
                headers=default_user_headers,
            )
            assert response.status_code == 200
            assert response.json()["id"] == default_slot_id
            if "time" in k:
                assert datetime.fromisoformat(response.json()[k]) == datetime.fromisoformat(v)
            else:
                assert response.json()[k] == v

    async def test_remove_template(
        self,
        client: AsyncClient,
        default_public_project: Project,
        default_user_headers: dict,
        default_user_org_membership,
        session: Session,
    ):
        response = await client.post(
            app.url_path_for("update_slot", project_id=default_project_id, slot_id=default_slot_id),
            json={"template_id": default_template_id},
            headers=default_user_headers,
        )
        assert response.status_code == 200
        slot = session.execute(
            select(Slot).where(Slot.project_id == default_project_id)
        ).scalar_one()
        assert slot.template_id == default_template_id

        response = await client.post(
            app.url_path_for("update_slot", project_id=default_project_id, slot_id=default_slot_id),
            json={"template_id": ""},
            headers=default_user_headers,
        )
        assert response.status_code == 200
        session.refresh(slot)
        assert slot.template_id is None

    async def test_bad_template_id(
        self,
        client: AsyncClient,
        default_public_project: Project,
        default_user_headers: dict,
        default_user_org_membership,
    ):
        path = app.url_path_for(
            "update_slot", project_id=default_project_id, slot_id=default_slot_id
        )
        response = await client.post(
            path, json={"template_id": "invalid uuid"}, headers=default_user_headers
        )
        assert response.status_code == 422
        response = await client.post(
            path, json={"template_id": str(uuid4())}, headers=default_user_headers
        )
        assert response.status_code == 400

    async def test_assign_template(
        self,
        client: AsyncClient,
        default_public_project: Project,
        default_user_headers: dict,
        default_user_org_membership,
        session: Session,
    ):
        path = app.url_path_for(
            "update_slot", project_id=default_project_id, slot_id=default_slot_id
        )
        template = SlotTemplate(project_id=default_project_id, title="template 0")
        session.add(template)
        session.commit()

        response = await client.post(
            path, json={"template_id": template.id}, headers=default_user_headers
        )
        assert response.status_code == 200
        assert response.json()["template_id"] == template.id
        session.refresh(template)
        assert len(template.slots) == 1

        response = await client.post(path, json={"template_id": None}, headers=default_user_headers)
        assert response.status_code == 200
        assert response.json()["template_id"] is None

    async def test_update_volunteers(
        self,
        client: AsyncClient,
        default_public_project: Project,
        default_user_headers: dict,
        default_user_org_membership,
        session: Session,
    ):
        path = app.url_path_for(
            "update_slot", project_id=default_project_id, slot_id=default_slot_id
        )

        response = await client.post(path, json={"volunteers": []}, headers=default_user_headers)
        assert response.status_code == 200
        volunteer = (
            session.execute(select(Volunteer).where(Volunteer.id == default_volunteer_id))
            .scalars()
            .first()
        )
        assert volunteer.slots_id == []

        response = await client.post(
            path, json={"volunteers": [default_volunteer_id]}, headers=default_user_headers
        )
        assert response.status_code == 200
        session.refresh(volunteer)
        assert volunteer.slots_id == [default_slot_id]

        response = await client.post(
            path, json={"volunteers": [str(uuid4())]}, headers=default_user_headers
        )
        assert response.status_code == 400

        response = await client.post(
            path, json={"volunteers": ["not uuid str"]}, headers=default_user_headers
        )
        assert response.status_code == 422

    async def test_respo_commission_updates_own_slot(
        self,
        client: AsyncClient,
        default_public_project: Project,
        commission_and_member,
        commission_template,
        session: Session,
    ):
        _, _, headers = commission_and_member
        slot = Slot(
            project_id=default_project_id,
            title="Bar shift",
            starting_time=datetime.now(),
            ending_time=datetime.now() + timedelta(hours=1),
            template_id=commission_template.id,
        )
        session.add(slot)
        session.commit()

        response = await client.post(
            app.url_path_for("update_slot", project_id=default_project_id, slot_id=slot.id),
            json={"title": "Bar shift renamed"},
            headers=headers,
        )
        assert response.status_code == 200
        session.refresh(slot)
        assert slot.title == "Bar shift renamed"

    async def test_respo_commission_rejected_for_other_commission_slot(
        self,
        client: AsyncClient,
        default_public_project: Project,
        commission_and_member,
        other_commission_template,
        session: Session,
    ):
        _, _, headers = commission_and_member
        slot = Slot(
            project_id=default_project_id,
            title="Scene shift",
            starting_time=datetime.now(),
            ending_time=datetime.now() + timedelta(hours=1),
            template_id=other_commission_template.id,
        )
        session.add(slot)
        session.commit()

        response = await client.post(
            app.url_path_for("update_slot", project_id=default_project_id, slot_id=slot.id),
            json={"title": "Hijacked"},
            headers=headers,
        )
        assert response.status_code == 403

    async def test_respo_commission_reassign_to_other_commission_template_rejected(
        self,
        client: AsyncClient,
        default_public_project: Project,
        commission_and_member,
        commission_template,
        other_commission_template,
        session: Session,
    ):
        _, _, headers = commission_and_member
        slot = Slot(
            project_id=default_project_id,
            title="Bar shift",
            starting_time=datetime.now(),
            ending_time=datetime.now() + timedelta(hours=1),
            template_id=commission_template.id,
        )
        session.add(slot)
        session.commit()

        response = await client.post(
            app.url_path_for("update_slot", project_id=default_project_id, slot_id=slot.id),
            json={"template_id": other_commission_template.id},
            headers=headers,
        )
        assert response.status_code == 403


class TestDeleteSlot:
    async def test_invalid_slot_id(
        self,
        client: AsyncClient,
        default_public_project: Project,
        default_user_headers: dict,
        default_user_org_membership,
    ):
        response = await client.delete(
            app.url_path_for("delete_slot", project_id=default_project_id, slot_id="not uidstr"),
            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_slot", project_id=default_project_id, slot_id=default_slot_id),
            headers=default_user_headers,
        )
        assert response.status_code == 200
        assert session.get(Slot, default_slot_id) is None

        volunteer = (
            session.execute(select(Volunteer).where(Volunteer.id == default_volunteer_id))
            .scalars()
            .first()
        )
        assert default_slot_id not in volunteer.slots_id

    async def test_nonexistent_slot_returns_404(
        self,
        client: AsyncClient,
        default_public_project: Project,
        default_user_headers: dict,
        default_user_org_membership,
    ):
        """require_commission_scope resolves the slot before the handler
        runs, so a missing slot_id now 404s -- differs from the old bulk
        DELETE behavior, which silently returned 200 either way."""
        response = await client.delete(
            app.url_path_for("delete_slot", project_id=default_project_id, slot_id=uuid4()),
            headers=default_user_headers,
        )
        assert response.status_code == 404

    async def test_respo_commission_deletes_own_slot(
        self,
        client: AsyncClient,
        default_public_project: Project,
        commission_and_member,
        commission_template,
        session: Session,
    ):
        _, _, headers = commission_and_member
        slot = Slot(
            project_id=default_project_id,
            title="Bar shift",
            starting_time=datetime.now(),
            ending_time=datetime.now() + timedelta(hours=1),
            template_id=commission_template.id,
        )
        session.add(slot)
        session.commit()
        slot_id = slot.id
        session.expunge(slot)
        response = await client.delete(
            app.url_path_for("delete_slot", project_id=default_project_id, slot_id=slot_id),
            headers=headers,
        )
        assert response.status_code == 200
        assert session.get(Slot, slot_id) is None

    async def test_respo_commission_rejected_for_other_commission_slot(
        self,
        client: AsyncClient,
        default_public_project: Project,
        commission_and_member,
        other_commission_template,
        session: Session,
    ):
        _, _, headers = commission_and_member
        slot = Slot(
            project_id=default_project_id,
            title="Scene shift",
            starting_time=datetime.now(),
            ending_time=datetime.now() + timedelta(hours=1),
            template_id=other_commission_template.id,
        )
        session.add(slot)
        session.commit()

        response = await client.delete(
            app.url_path_for("delete_slot", project_id=default_project_id, slot_id=slot.id),
            headers=headers,
        )
        assert response.status_code == 403
        assert session.get(Slot, slot.id) is not None