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

import uuid
from datetime import UTC, 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, Sms, Volunteer
from app.tests.conftest import default_project_id, default_sms_id, default_volunteer_id
from app.tests.shared_access import SharedProjectAccessTests

pytestmark = pytest.mark.asyncio

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

route_kwarg = {"project_id": default_project_id}
route_kwargs = {**route_kwarg, "sms_id": default_sms_id}
SMS_ROUTES = [
    ("GET", "list_project_sms", route_kwarg, None),
    (
        "POST",
        "create_sms",
        route_kwarg,
        {
            "phone_number": "Test",
            "content": "Coucou",
        },
    ),
    ("POST", "update_sms", route_kwargs, {"title": "Updated"}),
    ("DELETE", "delete_sms", route_kwargs, None),
]


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

        return request.param


class TestListProjectSms:
    async def test_invalid_project_id_format(self, client: AsyncClient, make_org_user):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        response = await client.get(
            app.url_path_for("list_project_sms", project_id="pas un uuid valid"), headers=headers
        )
        assert response.status_code == 422

    @pytest.mark.parametrize("role", ALL_ROLES)
    async def test_all_roles_can_read(
        self, client: AsyncClient, default_public_project: Project, make_org_user, role
    ):
        _, headers = make_org_user(role=role)
        response = await client.get(
            app.url_path_for("list_project_sms", project_id=default_project_id), headers=headers
        )
        assert response.status_code == 200
        data = response.json()
        assert len(data) == 1
        assert data[0]["id"] == default_sms_id


class TestCreateSms:
    @pytest.mark.parametrize("role", ALL_ROLES)
    async def test_all_roles_can_create(
        self,
        client: AsyncClient,
        default_public_project: Project,
        make_org_user,
        session: Session,
        role,
    ):
        _, headers = make_org_user(role=role)
        payload = {"phone_number": "06 75 75 75 75 ", "content": "sms_content"}

        before_creation_time = datetime.now(UTC)
        response = await client.post(
            app.url_path_for("create_sms", project_id=default_project_id),
            json=payload,
            headers=headers,
        )
        assert response.status_code == 200
        assert response.json()["content"] == "sms_content"

        sms = (
            session.execute(
                select(Sms).where(
                    (Sms.project_id == default_project_id) & (Sms.id != default_sms_id)
                )
            )
            .scalars()
            .first()
        )
        assert sms is not None
        assert sms.content == "sms_content"
        assert before_creation_time < sms.sending_time
        assert sms.sending_time < datetime.now(UTC)

    async def test_no_membership_forbidden(
        self, client: AsyncClient, default_public_project: Project, make_org_user
    ):
        _, headers = make_org_user(role=None)
        payload = {"phone_number": "06 75 75 75 75 ", "content": "sms_content"}
        response = await client.post(
            app.url_path_for("create_sms", project_id=default_project_id),
            json=payload,
            headers=headers,
        )
        assert response.status_code == 403

    async def test_invalid_payload(
        self, client: AsyncClient, default_public_project: Project, make_org_user
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        payload = {"phone_number": "06 75 75 75 75 "}  # missing content
        response = await client.post(
            app.url_path_for("create_sms", project_id=default_project_id),
            json=payload,
            headers=headers,
        )
        assert response.status_code == 422


class TestUpdateSms:
    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("update_sms", project_id=default_project_id, sms_id=default_sms_id),
            json={"volunteer_id": True},
            headers=headers,
        )
        assert response.status_code == 422

    async def test_sms_not_found(
        self, client: AsyncClient, default_public_project: Project, make_org_user
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        payload = {
            "volunteer_id": default_volunteer_id,
            "phone_number": "06 75 75 75 75 ",
            "content": "sms_content",
            "sending_time": datetime(2024, 5, 17, tzinfo=UTC).isoformat(),
        }
        response = await client.post(
            app.url_path_for("update_sms", project_id=default_project_id, sms_id=uuid.uuid4()),
            json=payload,
            headers=headers,
        )
        assert response.status_code == 404

    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("update_sms", project_id="pas un uuid valid", sms_id=default_sms_id),
            json={"content": "x"},
            headers=headers,
        )
        assert response.status_code == 422

    @pytest.mark.parametrize("role", ALL_ROLES)
    async def test_all_roles_can_update_each_field(
        self, client: AsyncClient, default_public_project: Project, make_org_user, role
    ):
        _, headers = make_org_user(role=role)
        payload = {
            "volunteer_id": default_volunteer_id,
            "phone_number": "06 75 75 75 75 ",
            "content": "sms_content",
            "sending_time": datetime(2024, 5, 17, tzinfo=UTC).isoformat(),
        }
        for k, v in payload.items():
            response = await client.post(
                app.url_path_for(
                    "update_sms", project_id=default_project_id, sms_id=default_sms_id
                ),
                json={k: v},
                headers=headers,
            )
            assert response.status_code == 200
            assert response.json()["id"] == default_sms_id
            if "time" in k:
                parsed_time = datetime.fromisoformat(response.json()[k])
                assert parsed_time == datetime.fromisoformat(v)
            else:
                assert response.json()[k] == v

    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.post(
            app.url_path_for("update_sms", project_id=default_project_id, sms_id=default_sms_id),
            json={"content": "hijacked"},
            headers=headers,
        )
        assert response.status_code == 403


class TestDeleteSms:
    async def test_invalid_sms_id_format(
        self, client: AsyncClient, default_public_project: Project, make_org_user
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        response = await client.delete(
            app.url_path_for("delete_sms", project_id=default_project_id, sms_id="not uidstr"),
            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)
        response = await client.delete(
            app.url_path_for("delete_sms", project_id=default_project_id, sms_id=default_sms_id),
            headers=headers,
        )
        assert response.status_code == 403
        assert session.get(Sms, default_sms_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)
        response = await client.delete(
            app.url_path_for("delete_sms", project_id=default_project_id, sms_id=default_sms_id),
            headers=headers,
        )
        assert response.status_code == 200
        assert session.get(Sms, default_sms_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_sms", project_id=default_project_id, sms_id=default_sms_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
    ):
        """sms.py still uses bulk DELETE without an existence check --
        matches volunteers.py's behavior, not slots.py/templates.py's 404.
        Same cross-router inconsistency flagged for delete_volunteer."""
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        response = await client.delete(
            app.url_path_for("delete_sms", project_id=default_project_id, sms_id=uuid.uuid4()),
            headers=headers,
        )
        assert response.status_code == 200

    async def test_delete_cascades_from_volunteer(
        self, client: AsyncClient, default_public_project: Project, make_org_user, session: Session
    ):
        _, headers = make_org_user(role=OrgRole.ORG_ADMIN)
        volunteer = (
            session.execute(select(Volunteer).where(Volunteer.id == default_volunteer_id))
            .scalars()
            .first()
        )
        assert len(volunteer.sms) == 0

        sms = session.get(Sms, default_sms_id)
        sms.volunteer_id = default_volunteer_id
        session.commit()
        session.refresh(volunteer)
        assert len(volunteer.sms) == 1, "Sms not added to volunteer"
        session.expunge(sms)

        response = await client.delete(
            app.url_path_for("delete_sms", project_id=default_project_id, sms_id=default_sms_id),
            headers=headers,
        )
        assert response.status_code == 200
        assert session.get(Sms, default_sms_id) is None

        session.refresh(volunteer)
        assert len(volunteer.sms) == 0, "Sms should be deleted"

        volunteer_after = (
            session.execute(select(Volunteer).where(Volunteer.id == default_volunteer_id))
            .scalars()
            .first()
        )
        assert volunteer_after is not None, "Volunteer should not be deleted"
        assert len(volunteer_after.sms) == 0