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

from datetime import datetime, timedelta, 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, Slot, SlotTemplate, Volunteer
from app.tests.conftest import (
    default_project_id,
    default_slot_id,
    default_volunteer_id,
    default_template_id,
)


async def test_read_list_project_slots(
    client: AsyncClient, default_user_headers: dict, default_public_project: Project
):
    response = await client.get(
        app.url_path_for("list_project_slots", project_id=default_project_id),
    )
    assert response.status_code == 401
    response = await client.get(
        app.url_path_for("list_project_slots", project_id=uuid.uuid4()),
        headers=default_user_headers,
    )
    assert response.status_code == 404
    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

    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
    slot_response = data[0]
    assert slot_response["title"] == "être roi"
    assert slot_response["id"] == default_slot_id
    assert "created_at" in slot_response


async def test_create_slot(
    client: AsyncClient,
    default_public_project: Project,
    default_user_headers: dict,
    session: Session,
):
    # Test without autentication
    response = await client.post(app.url_path_for("create_slot", project_id=default_project_id))
    assert response.status_code == 401
    starting_time = datetime(1900, 1, 1)
    payload = {
        "title": "être mort",
        "starting_time": starting_time.isoformat(),
        "ending_time": (starting_time + timedelta(minutes=60)).isoformat(),
    }
    # Test invalid project_id
    response = await client.post(
        app.url_path_for("create_slot", project_id=uuid.uuid4()),
        json=payload,
        headers=default_user_headers,
    )
    assert response.status_code == 404

    # Test normal payload
    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 slot.required_volunteers == 0
    assert abs(slot.starting_time - starting_time.replace(tzinfo=timezone.utc)) < timedelta(
        minutes=30
    )

    # Test invalid payload
    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_create_slot_min_volunteer(
    client: AsyncClient,
    default_public_project: Project,
    default_user_headers: dict,
    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,
    }
    # Test normal payload
    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

    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.required_volunteers == 2


async def test_update_slot(
    client: AsyncClient,
    default_public_project: Project,
    default_user_headers: dict,
    session: Session,
):
    # Test without autentication
    response = await client.post(
        app.url_path_for(
            "update_slot",
            project_id=default_project_id,
            slot_id=default_slot_id,
        )
    )
    assert response.status_code == 401

    starting_time = datetime(2000, 1, 1, tzinfo=timezone.utc)
    payload = {
        "title": "être mort 2 fois",
        "starting_time": starting_time.isoformat(),
        "ending_time": (starting_time + timedelta(minutes=60)).isoformat(),
        "required_volunteers": 2,
    }

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

    # test invalid slot_id
    response = await client.post(
        app.url_path_for(
            "update_slot",
            project_id=default_project_id,
            slot_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_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_update_slot_remove_template(
    client: AsyncClient,
    default_public_project: Project,
    default_user_headers: dict,
    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_update_bad_slot_template(
    client: AsyncClient,
    default_public_project: Project,
    default_user_headers: dict,
):
    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(uuid.uuid4())},
        headers=default_user_headers,
    )
    assert response.status_code == 400


async def test_update_slot_template(
    client: AsyncClient,
    default_public_project: Project,
    default_user_headers: dict,
    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_slot_volunteers(
    client: AsyncClient,
    default_public_project: Project,
    default_user_headers: dict,
    session: Session,
):
    response = await client.post(
        app.url_path_for(
            "update_slot",
            project_id=default_project_id,
            slot_id=default_slot_id,
        ),
        json={"volunteers": []},
        headers=default_user_headers,
    )

    assert response.status_code == 200
    result = session.execute(select(Volunteer).where(Volunteer.id == default_volunteer_id))
    volunteer = result.scalars().first()
    assert volunteer is not None
    assert volunteer.slots_id == []

    response = await client.post(
        app.url_path_for(
            "update_slot",
            project_id=default_project_id,
            slot_id=default_slot_id,
        ),
        json={"volunteers": [default_volunteer_id]},
        headers=default_user_headers,
    )

    assert response.status_code == 200
    session.refresh(volunteer)
    assert volunteer is not None
    assert volunteer.slots_id == [default_slot_id]

    # An invalid volunteer list
    response = await client.post(
        app.url_path_for(
            "update_slot",
            project_id=default_project_id,
            slot_id=default_slot_id,
        ),
        json={"volunteers": [str(uuid.uuid4())]},
        headers=default_user_headers,
    )

    assert response.status_code == 400

    # An invalid volunteer list
    response = await client.post(
        app.url_path_for(
            "update_slot",
            project_id=default_project_id,
            slot_id=default_slot_id,
        ),
        json={"volunteers": ["not uuid str"]},
        headers=default_user_headers,
    )

    assert response.status_code == 422


async def test_delete_slot_fail(
    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_slot",
            project_id=default_project_id,
            slot_id=default_slot_id,
        )
    )
    assert response.status_code == 401
    result = session.execute(select(Slot).where(Slot.id == default_slot_id))
    slot = result.scalars().first()
    assert slot is not None

    # Cannot delete non uuid string
    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_delete_slot(
    client: AsyncClient,
    default_user_headers: dict,
    session: Session,
    default_public_project: Project,
):
    # Proper deletion
    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
    result = session.execute(select(Slot).where(Slot.id == default_slot_id))
    slot = result.scalars().first()
    assert slot is None

    # check deletion is cascaded to volunteers
    result = session.execute(select(Volunteer).where(Volunteer.id == default_volunteer_id))
    volunteer: Volunteer | None = result.scalars().first()
    assert volunteer is not None
    assert default_slot_id not in volunteer.slots_id

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


async def test_delete_slot_idempotent(
    client: AsyncClient,
    default_user_headers: dict,
    session: Session,
    default_public_project: Project,
):
    # Idempotence test
    url = app.url_path_for("delete_slot", project_id=default_project_id, slot_id=default_slot_id)
    response = await client.delete(url, headers=default_user_headers)
    response = await client.delete(url, headers=default_user_headers)
    assert response.status_code == 200