|
|
@@ -0,0 +1,300 @@
|
|
|
+from datetime import datetime, timezone
|
|
|
+from re import template
|
|
|
+import uuid
|
|
|
+from httpx import AsyncClient
|
|
|
+import pytest
|
|
|
+from sqlalchemy import select
|
|
|
+from sqlalchemy.orm import Session
|
|
|
+
|
|
|
+from app.main import app
|
|
|
+from app.models import Project, Slot, SlotTag, SlotTemplate
|
|
|
+from app.tests.conftest import default_project_id, default_tag_id
|
|
|
+
|
|
|
+
|
|
|
+async def test_read_list_project_tags(
|
|
|
+ client: AsyncClient,
|
|
|
+ default_user_headers: dict,
|
|
|
+ default_public_project: Project,
|
|
|
+ session: Session,
|
|
|
+):
|
|
|
+ response = await client.get(
|
|
|
+ app.url_path_for("list_project_tags", project_id=default_project_id),
|
|
|
+ )
|
|
|
+ assert response.status_code == 401
|
|
|
+ response = await client.get(
|
|
|
+ app.url_path_for("list_project_tags", project_id=default_project_id),
|
|
|
+ headers=default_user_headers,
|
|
|
+ )
|
|
|
+ assert response.status_code == 200
|
|
|
+ assert len(response.json()) == 1
|
|
|
+ tag = SlotTag(title="1er tag", project_id=default_project_id)
|
|
|
+ session.add(tag)
|
|
|
+ session.commit()
|
|
|
+ response = await client.get(
|
|
|
+ app.url_path_for("list_project_tags", project_id=default_project_id),
|
|
|
+ headers=default_user_headers,
|
|
|
+ )
|
|
|
+ assert response.status_code == 200
|
|
|
+ assert len(response.json()) == 2
|
|
|
+
|
|
|
+
|
|
|
+async def test_create_tag_fail(
|
|
|
+ client: AsyncClient,
|
|
|
+ default_user_headers: dict,
|
|
|
+ default_public_project: Project,
|
|
|
+ session: Session,
|
|
|
+):
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_tag", project_id="default_project_id"),
|
|
|
+ headers=default_user_headers,
|
|
|
+ )
|
|
|
+ assert response.status_code == 422
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_tag", project_id=uuid.uuid4()),
|
|
|
+ json={"title": "1st tag"},
|
|
|
+ headers=default_user_headers,
|
|
|
+ )
|
|
|
+ assert response.status_code == 404
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_tag", project_id="default_project_id"),
|
|
|
+ )
|
|
|
+ assert response.status_code == 401
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_tag", project_id=default_project_id),
|
|
|
+ json={"titl": "1st tag"},
|
|
|
+ headers=default_user_headers,
|
|
|
+ )
|
|
|
+ assert response.status_code == 422
|
|
|
+
|
|
|
+
|
|
|
+async def test_create_tag(
|
|
|
+ client: AsyncClient,
|
|
|
+ default_user_headers: dict,
|
|
|
+ default_project: Project,
|
|
|
+ session: Session,
|
|
|
+):
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_tag", project_id=default_project_id),
|
|
|
+ json={"title": "1st tag"},
|
|
|
+ headers=default_user_headers,
|
|
|
+ )
|
|
|
+ assert response.status_code == 200
|
|
|
+ result = session.execute(select(SlotTag).where(SlotTag.project_id == default_project_id))
|
|
|
+ tag = result.scalars().first()
|
|
|
+ assert tag is not None
|
|
|
+ assert tag.title == "1st tag"
|
|
|
+
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_tag", project_id=default_project_id),
|
|
|
+ json={"title": "1st tag"},
|
|
|
+ headers=default_user_headers,
|
|
|
+ )
|
|
|
+ result = session.execute(select(SlotTag).where(SlotTag.project_id == default_project_id))
|
|
|
+ tag_count = 0
|
|
|
+ for tag in result.scalars():
|
|
|
+ tag_count += 1
|
|
|
+ assert tag_count == 2
|
|
|
+
|
|
|
+
|
|
|
+async def test_create_tag_with_template(
|
|
|
+ client: AsyncClient,
|
|
|
+ default_user_headers: dict,
|
|
|
+ default_public_project: Project,
|
|
|
+ session: Session,
|
|
|
+):
|
|
|
+ template = SlotTemplate(
|
|
|
+ project_id=default_project_id, title="coucou", description="ceci est une description"
|
|
|
+ )
|
|
|
+ session.add(template)
|
|
|
+ session.commit()
|
|
|
+
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_tag", project_id=default_project_id),
|
|
|
+ json={"title": "1st tag", "templates": [template.id]},
|
|
|
+ headers=default_user_headers,
|
|
|
+ )
|
|
|
+ assert response.status_code == 200
|
|
|
+ assert response.json()["templates_id"][0] == template.id
|
|
|
+ session.refresh(template)
|
|
|
+ assert len(template.tags) > 0
|
|
|
+
|
|
|
+ template = SlotTemplate(
|
|
|
+ project_id=default_project_id, title="coucou", description="ceci est une description"
|
|
|
+ )
|
|
|
+ session.add(template)
|
|
|
+ session.commit()
|
|
|
+
|
|
|
+ project = Project(name="second project", is_public=False)
|
|
|
+ session.add(project)
|
|
|
+ session.commit()
|
|
|
+ session.refresh(project)
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_tag", project_id=project.id),
|
|
|
+ json={"title": "1st tag", "templates": [template.id]},
|
|
|
+ headers=default_user_headers,
|
|
|
+ )
|
|
|
+ assert response.status_code == 400
|
|
|
+
|
|
|
+
|
|
|
+async def test_update_tag_fail(
|
|
|
+ client: AsyncClient,
|
|
|
+ default_user_headers: dict,
|
|
|
+ default_public_project: Project,
|
|
|
+ session: Session,
|
|
|
+):
|
|
|
+ # no authent
|
|
|
+ url = app.url_path_for("update_tag", project_id=default_project_id, tag_id=default_tag_id)
|
|
|
+ response = await client.post(url, json={"title": "royaux"})
|
|
|
+ assert response.status_code == 401
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.parametrize(
|
|
|
+ "payload", [{"title": [1.001, 2]}, {"title": {}}, {"templates": ["1", "2"]}]
|
|
|
+)
|
|
|
+async def test_update_tag_fail_invalid_payload(
|
|
|
+ client: AsyncClient,
|
|
|
+ default_user_headers: dict,
|
|
|
+ default_public_project: Project,
|
|
|
+ session: Session,
|
|
|
+ payload: dict,
|
|
|
+):
|
|
|
+ url = app.url_path_for("update_tag", project_id=default_project_id, tag_id=default_tag_id)
|
|
|
+ response = await client.post(url, json=payload, headers=default_user_headers)
|
|
|
+ assert response.status_code == 422
|
|
|
+
|
|
|
+
|
|
|
+async def test_update_tag(
|
|
|
+ client: AsyncClient,
|
|
|
+ default_user_headers: dict,
|
|
|
+ default_public_project: Project,
|
|
|
+ session: Session,
|
|
|
+):
|
|
|
+ tag = session.get(SlotTag, default_tag_id)
|
|
|
+ assert tag is not None
|
|
|
+ url = app.url_path_for("update_tag", project_id=default_project_id, tag_id=default_tag_id)
|
|
|
+
|
|
|
+ response = await client.post(url, json={"title": "new_title"}, headers=default_user_headers)
|
|
|
+ assert response.status_code == 200
|
|
|
+ session.refresh(tag)
|
|
|
+ assert tag.title == "new_title"
|
|
|
+ assert response.json()["title"] == "new_title"
|
|
|
+
|
|
|
+ template = SlotTemplate(project_id=default_project_id, title="template")
|
|
|
+ session.add(template)
|
|
|
+ session.commit()
|
|
|
+ response = await client.post(
|
|
|
+ url, json={"templates": [template.id]}, headers=default_user_headers
|
|
|
+ )
|
|
|
+ assert response.status_code == 200
|
|
|
+ session.refresh(tag)
|
|
|
+ assert len(tag.templates) == 1
|
|
|
+ response = await client.post(url, json={"templates": []}, headers=default_user_headers)
|
|
|
+ session.refresh(tag)
|
|
|
+ assert len(tag.templates) == 0
|
|
|
+
|
|
|
+
|
|
|
+async def test_update_tag_fail_template(
|
|
|
+ client: AsyncClient,
|
|
|
+ default_user_headers: dict,
|
|
|
+ default_public_project: Project,
|
|
|
+ session: Session,
|
|
|
+):
|
|
|
+ tag = session.get(SlotTag, default_tag_id)
|
|
|
+ assert tag is not None
|
|
|
+ project = Project(name="other project", is_public=False)
|
|
|
+ template = SlotTemplate(title="template 1")
|
|
|
+ project.templates.append(template)
|
|
|
+ session.add(template)
|
|
|
+ session.add(project)
|
|
|
+ session.commit()
|
|
|
+ url = app.url_path_for("update_tag", project_id=default_project_id, tag_id=default_tag_id)
|
|
|
+ response = await client.post(
|
|
|
+ url, json={"templates": [template.id]}, headers=default_user_headers
|
|
|
+ )
|
|
|
+ assert response.status_code == 400
|
|
|
+
|
|
|
+
|
|
|
+@pytest.mark.parametrize("n_slot", [1, 5, 10])
|
|
|
+async def test_list_slot(
|
|
|
+ client: AsyncClient,
|
|
|
+ default_user_headers: dict,
|
|
|
+ session: Session,
|
|
|
+ default_public_project: Project,
|
|
|
+ n_slot: int,
|
|
|
+):
|
|
|
+ url = app.url_path_for("list_tagged_slot", project_id=default_project_id, tag_id=default_tag_id)
|
|
|
+ response = await client.get(url, headers=default_user_headers)
|
|
|
+ assert response.status_code == 200
|
|
|
+ assert len(response.json()) == 0
|
|
|
+ tag = session.get(SlotTag, default_tag_id)
|
|
|
+ assert tag is not None
|
|
|
+ for i in range(n_slot):
|
|
|
+ template = SlotTemplate(project_id=default_project_id, title=f"template {i}")
|
|
|
+ slot = Slot(
|
|
|
+ project_id=default_project_id,
|
|
|
+ title=f"Slot {i}",
|
|
|
+ starting_time=datetime(2024, 9, 9, 12, 2 * i),
|
|
|
+ ending_time=datetime(2024, 9, 9, 12, 2 * (i + 1)),
|
|
|
+ )
|
|
|
+ slot.template = template
|
|
|
+ tag.templates.append(template)
|
|
|
+ session.add_all([template, slot])
|
|
|
+
|
|
|
+ session.commit()
|
|
|
+ response = await client.get(url, headers=default_user_headers)
|
|
|
+ assert response.status_code == 200
|
|
|
+ assert len(response.json()) == n_slot
|
|
|
+
|
|
|
+
|
|
|
+async def test_delete_tag_fail(
|
|
|
+ client: AsyncClient,
|
|
|
+ default_user_headers: dict,
|
|
|
+ session: Session,
|
|
|
+ default_public_project: Project,
|
|
|
+):
|
|
|
+ # no authent
|
|
|
+ url = app.url_path_for("delete_tag", project_id=default_project_id, tag_id=default_tag_id)
|
|
|
+ response = await client.delete(url)
|
|
|
+ assert response.status_code == 401
|
|
|
+ # invalid tag
|
|
|
+ url = app.url_path_for("delete_tag", project_id=default_project_id, tag_id="default_tag_id")
|
|
|
+ response = await client.delete(url, headers=default_user_headers)
|
|
|
+ assert response.status_code == 422
|
|
|
+
|
|
|
+ # invalid project_id
|
|
|
+ url = app.url_path_for("delete_tag", project_id="default_volunteer_id", tag_id=default_tag_id)
|
|
|
+ response = await client.delete(url, headers=default_user_headers)
|
|
|
+ assert response.status_code == 422
|
|
|
+
|
|
|
+
|
|
|
+async def test_delete_tag(
|
|
|
+ client: AsyncClient,
|
|
|
+ default_user_headers: dict,
|
|
|
+ session: Session,
|
|
|
+ default_public_project: Project,
|
|
|
+):
|
|
|
+ # Proper deletion
|
|
|
+ url = app.url_path_for("delete_tag", project_id=default_project_id, tag_id=default_tag_id)
|
|
|
+ response = await client.delete(url, headers=default_user_headers)
|
|
|
+ assert response.status_code == 200
|
|
|
+ result = session.execute(select(SlotTag).where(SlotTag.id == default_tag_id))
|
|
|
+ slot = result.scalars().first()
|
|
|
+ assert slot is None
|
|
|
+
|
|
|
+ # can delete random uuid
|
|
|
+ url = app.url_path_for("delete_tag", project_id=default_project_id, tag_id=uuid.uuid4())
|
|
|
+ response = await client.delete(url, headers=default_user_headers)
|
|
|
+ assert response.status_code == 200
|
|
|
+
|
|
|
+
|
|
|
+async def test_delete_tag_idempotent(
|
|
|
+ client: AsyncClient,
|
|
|
+ default_user_headers: dict,
|
|
|
+ session: Session,
|
|
|
+ default_public_project: Project,
|
|
|
+):
|
|
|
+ # Idempotence test
|
|
|
+ url = app.url_path_for("delete_tag", project_id=default_project_id, tag_id=default_tag_id)
|
|
|
+ response = await client.delete(url, headers=default_user_headers)
|
|
|
+ response = await client.delete(url, headers=default_user_headers)
|
|
|
+ assert response.status_code == 200
|