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

from datetime import datetime, timezone
import uuid
from httpx import AsyncClient
from sqlalchemy import select
from sqlalchemy.orm import Session

from app.main import app
from app.models import Project, Sms, Volunteer
from app.tests.conftest import default_project_id, default_volunteer_id, default_sms_id


async def test_read_list_project_sms(
    client: AsyncClient, default_user_headers: dict, default_public_project: Project
):
    response = await client.get(
        app.url_path_for("list_project_sms", project_id=default_project_id),
    )
    assert response.status_code == 401
    response = await client.get(
        app.url_path_for("list_project_sms", project_id=uuid.uuid4()),
        headers=default_user_headers,
    )
    assert response.status_code == 404
    response = await client.get(
        app.url_path_for("list_project_sms", project_id="pas un uuid valid"),
        headers=default_user_headers,
    )
    assert response.status_code == 422

    response = await client.get(
        app.url_path_for("list_project_sms", project_id=default_project_id),
        headers=default_user_headers,
    )
    assert response.status_code == 200
    data = response.json()
    assert len(data) == 1
    assert data[0]["id"] == default_sms_id


async def test_create_sms(
    client: AsyncClient,
    default_public_project: Project,
    default_user_headers: dict,
    session: Session,
):
    # Test without autentication
    response = await client.post(app.url_path_for("create_sms", project_id=default_project_id))
    assert response.status_code == 401
    payload = {"phone_number": "06 75 75 75 75 ", "content": "sms_content"}
    # test invalid project_id
    response = await client.post(
        app.url_path_for("create_sms", project_id=uuid.uuid4()),
        json=payload,
        headers=default_user_headers,
    )
    assert response.status_code == 404

    # Test normal payload
    before_creation_time = datetime.now(timezone.utc)
    response = await client.post(
        app.url_path_for("create_sms", project_id=default_project_id),
        json=payload,
        headers=default_user_headers,
    )
    assert response.status_code == 200
    assert response.json()["content"] == "sms_content"
    result = session.execute(
        select(Sms).where((Sms.project_id == default_project_id) & (Sms.id != default_sms_id))
    )
    sms = result.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(timezone.utc)

    # test invalid payload
    del payload["content"]
    response = await client.post(
        app.url_path_for("create_sms", project_id=default_project_id),
        json=payload,
        headers=default_user_headers,
    )
    assert response.status_code == 422


async def test_update_sms(
    client: AsyncClient,
    default_public_project: Project,
    default_user_headers: dict,
    session: Session,
):
    # Test without autentication
    response = await client.post(
        app.url_path_for(
            "update_sms",
            project_id=default_project_id,
            sms_id=default_sms_id,
        )
    )
    assert response.status_code == 401

    # Test invalid payload
    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=default_user_headers,
    )
    assert response.status_code == 422

    payload = {
        "volunteer_id": default_volunteer_id,
        "phone_number": "06 75 75 75 75 ",
        "content": "sms_content",
        "sending_time": datetime(2024, 5, 17, tzinfo=timezone.utc).isoformat(),
    }

    # test invalid project_id
    response = await client.post(
        app.url_path_for(
            "update_sms",
            project_id=uuid.uuid4(),
            sms_id=default_sms_id,
        ),
        json=payload,
        headers=default_user_headers,
    )
    assert response.status_code == 404

    # test invalid sms_id
    response = await client.post(
        app.url_path_for(
            "update_sms",
            project_id=default_project_id,
            sms_id=uuid.uuid4(),
        ),
        json=payload,
        headers=default_user_headers,
    )
    assert response.status_code == 404

    # Test normal payload
    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=default_user_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_delete_sms(
    client: AsyncClient,
    default_user_headers: dict,
    session: Session,
    default_public_project: Project,
):
    # Fail deleting the project due to not logged in
    response = await client.delete(
        app.url_path_for(
            "delete_sms",
            project_id=default_project_id,
            sms_id=default_sms_id,
        )
    )
    assert response.status_code == 401
    result = session.execute(select(Sms).where(Sms.id == default_sms_id))
    sms = result.scalars().first()
    assert sms is not None

    # Proper deletion
    response = await client.delete(
        app.url_path_for(
            "delete_sms",
            project_id=default_project_id,
            sms_id=default_sms_id,
        ),
        headers=default_user_headers,
    )
    assert response.status_code == 200
    result = session.execute(select(Sms).where(Sms.id == default_sms_id))
    sms = result.scalars().first()
    assert sms is None

    # Idempotence test
    response = await client.delete(
        app.url_path_for(
            "delete_sms",
            project_id=default_project_id,
            sms_id=default_sms_id,
        ),
        headers=default_user_headers,
    )
    assert response.status_code == 200

    # can delete random uuid
    response = await client.delete(
        app.url_path_for("delete_sms", project_id=default_project_id, sms_id=uuid.uuid4()),
        headers=default_user_headers,
    )
    assert response.status_code == 200

    # Cannot delete non uuid string
    response = await client.delete(
        app.url_path_for("delete_sms", project_id=default_project_id, sms_id="not uidstr"),
        headers=default_user_headers,
    )
    assert response.status_code == 422


async def test_delete_sms_with_volunteer(
    client: AsyncClient,
    default_user_headers: dict,
    session: Session,
    default_public_project: Project,
):
    result = session.execute(select(Volunteer).where(Volunteer.id == default_volunteer_id))
    volunteer = result.scalars().first()
    assert volunteer is not None
    assert len(volunteer.sms) == 0

    result = session.execute(select(Sms).where(Sms.id == default_sms_id))
    sms = result.scalars().first()
    assert sms is not None
    sms.volunteer_id = default_volunteer_id
    session.commit()

    session.refresh(volunteer)
    assert len(volunteer.sms) == 1, "Sms not added to volunteer"

    response = await client.delete(
        app.url_path_for(
            "delete_sms",
            project_id=default_project_id,
            sms_id=default_sms_id,
        ),
        headers=default_user_headers,
    )
    assert response.status_code == 200
    result = session.execute(select(Sms).where(Sms.id == default_sms_id))
    assert result.scalars().first() is None

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

    result = session.execute(select(Volunteer).where(Volunteer.id == default_volunteer_id))
    volunteer_result = result.scalars().first()
    assert volunteer_result is not None, "Volunteer should not be deleted"
    assert len(volunteer_result.sms) == 0, "No more sms associated to volunteer"