|
|
@@ -0,0 +1,280 @@
|
|
|
+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, Sms, Volunteer
|
|
|
+from app.tests.conftest import default_project_id, default_project_name, default_sms_id
|
|
|
+
|
|
|
+
|
|
|
+async def test_read_list_project(
|
|
|
+ client: AsyncClient, default_user_headers: dict, default_project: Project
|
|
|
+):
|
|
|
+ response = await client.get(
|
|
|
+ app.url_path_for("list_project"), headers=default_user_headers
|
|
|
+ )
|
|
|
+ assert response.status_code == 200
|
|
|
+ data = response.json()
|
|
|
+ assert len(data) == 1
|
|
|
+ project_response = data[0]
|
|
|
+ assert project_response["id"] == default_project_id
|
|
|
+ assert "created_at" in project_response
|
|
|
+
|
|
|
+
|
|
|
+async def test_read_list_project_without_user(client: AsyncClient):
|
|
|
+ response = await client.get(app.url_path_for("list_project"))
|
|
|
+ assert response.status_code == 401
|
|
|
+
|
|
|
+
|
|
|
+async def test_read_list_public_project_without_user(
|
|
|
+ client: AsyncClient, default_project: Project
|
|
|
+):
|
|
|
+ response = await client.get(app.url_path_for("list_public_project"))
|
|
|
+ assert response.status_code == 200
|
|
|
+ data = response.json()
|
|
|
+ assert len(data) == 0
|
|
|
+
|
|
|
+
|
|
|
+async def test_read_list_public_project_without_user_1(
|
|
|
+ client: AsyncClient, default_public_project: Project
|
|
|
+):
|
|
|
+ response = await client.get(app.url_path_for("list_public_project"))
|
|
|
+ assert response.status_code == 200
|
|
|
+ data = response.json()
|
|
|
+ assert len(data) == 1
|
|
|
+
|
|
|
+
|
|
|
+async def test_get_project_without_user(client: AsyncClient, default_project: Project):
|
|
|
+ response = await client.get(
|
|
|
+ app.url_path_for("get_public_project", project_id=default_project_id)
|
|
|
+ )
|
|
|
+ assert response.status_code == 404
|
|
|
+
|
|
|
+
|
|
|
+async def test_get_public_project(client: AsyncClient, default_public_project: Project):
|
|
|
+ response = await client.get(
|
|
|
+ app.url_path_for("get_public_project", project_id=default_project_id)
|
|
|
+ )
|
|
|
+ assert response.status_code == 200
|
|
|
+ assert response.json()["id"] == default_project_id
|
|
|
+ assert response.json()["name"] == default_project_name
|
|
|
+
|
|
|
+
|
|
|
+async def test_get_project(
|
|
|
+ client: AsyncClient, default_public_project: Project, default_user_headers: dict
|
|
|
+):
|
|
|
+ # Test without autentication
|
|
|
+ response = await client.get(
|
|
|
+ app.url_path_for("get_project", project_id=default_project_id)
|
|
|
+ )
|
|
|
+ assert response.status_code == 401
|
|
|
+
|
|
|
+ response = await client.get(
|
|
|
+ app.url_path_for("get_project", project_id=default_project_id),
|
|
|
+ headers=default_user_headers,
|
|
|
+ )
|
|
|
+ assert response.status_code == 200
|
|
|
+ assert response.json()["id"] == default_project_id
|
|
|
+ assert response.json()["name"] == default_project_name
|
|
|
+
|
|
|
+
|
|
|
+async def test_create_project_without_user(client: AsyncClient, session: Session):
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_project"),
|
|
|
+ json={"name": "Coucou"},
|
|
|
+ )
|
|
|
+ assert response.status_code == 401
|
|
|
+ # Verify no project were added
|
|
|
+ result = session.execute(select(Project))
|
|
|
+ project = result.scalars().first()
|
|
|
+ assert project is None
|
|
|
+
|
|
|
+
|
|
|
+async def test_create_project(
|
|
|
+ client: AsyncClient, default_user_headers: dict, session: Session
|
|
|
+):
|
|
|
+ # Create a private project
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_project"),
|
|
|
+ headers=default_user_headers,
|
|
|
+ json={"name": "Coucou"},
|
|
|
+ )
|
|
|
+ assert response.status_code == 200
|
|
|
+ result = session.execute(select(Project).where(Project.name == "Coucou"))
|
|
|
+ project = result.scalars().first()
|
|
|
+ assert project is not None
|
|
|
+ assert project.is_public == False
|
|
|
+
|
|
|
+ # Create a public project
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_project"),
|
|
|
+ headers=default_user_headers,
|
|
|
+ json={"name": "Public", "is_public": True},
|
|
|
+ )
|
|
|
+ assert response.status_code == 200
|
|
|
+ result = session.execute(select(Project).where(Project.name == "Public"))
|
|
|
+ project = result.scalars().first()
|
|
|
+ assert project is not None
|
|
|
+ assert project.is_public
|
|
|
+
|
|
|
+
|
|
|
+async def test_create_project_fail(
|
|
|
+ client: AsyncClient, default_user_headers: dict, session: Session
|
|
|
+):
|
|
|
+ # Create a project with a bad
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_project"),
|
|
|
+ headers=default_user_headers,
|
|
|
+ json={"coucou": "Coucou"},
|
|
|
+ )
|
|
|
+ assert response.status_code == 422
|
|
|
+ assert "detail" in response.json()
|
|
|
+ result = session.execute(select(Project))
|
|
|
+ project = result.scalars().first()
|
|
|
+ assert project is None
|
|
|
+
|
|
|
+
|
|
|
+async def test_update_project(
|
|
|
+ client: AsyncClient,
|
|
|
+ default_user_headers: dict,
|
|
|
+ session: Session,
|
|
|
+ default_public_project: Project,
|
|
|
+):
|
|
|
+ # Update a public project
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("update_project", project_id=default_project_id),
|
|
|
+ headers=default_user_headers,
|
|
|
+ json={"name": "Coucou"},
|
|
|
+ )
|
|
|
+ assert response.status_code == 200
|
|
|
+ result = session.execute(select(Project).where(Project.id == default_project_id))
|
|
|
+ project = result.scalars().first()
|
|
|
+ assert project is not None
|
|
|
+ assert project.name == "Coucou"
|
|
|
+
|
|
|
+ # Update a public project
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("update_project", project_id=default_project_id),
|
|
|
+ headers=default_user_headers,
|
|
|
+ json={"name": "Coucou 2", "is_public": False},
|
|
|
+ )
|
|
|
+ assert response.status_code == 200
|
|
|
+ session.refresh(project)
|
|
|
+ assert project is not None
|
|
|
+ assert project.name == "Coucou 2"
|
|
|
+ assert project.created_at < project.updated_at
|
|
|
+ assert not project.is_public
|
|
|
+
|
|
|
+
|
|
|
+async def test_update_project_fail(
|
|
|
+ client: AsyncClient,
|
|
|
+ default_user_headers: dict,
|
|
|
+ session: Session,
|
|
|
+ default_public_project: Project,
|
|
|
+):
|
|
|
+ # Fail updating the project due to not logged in
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("update_project", project_id=default_project_id),
|
|
|
+ json={"name": "Coucou 2"},
|
|
|
+ )
|
|
|
+ assert response.status_code == 401
|
|
|
+ result = session.execute(select(Project).where(Project.id == default_project_id))
|
|
|
+ project = result.scalars().first()
|
|
|
+ assert project is not None
|
|
|
+ assert project.name == default_project_name
|
|
|
+
|
|
|
+ # Update is_public without name => Validation error
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("update_project", project_id=default_project_id),
|
|
|
+ headers=default_user_headers,
|
|
|
+ json={"is_public": False},
|
|
|
+ )
|
|
|
+ assert response.status_code == 422
|
|
|
+
|
|
|
+
|
|
|
+async def test_create_sms_batch(
|
|
|
+ client: AsyncClient,
|
|
|
+ default_user_headers: dict,
|
|
|
+ session: Session,
|
|
|
+ default_public_project: Project,
|
|
|
+):
|
|
|
+ #
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_sms_batch", project_id=default_project_id),
|
|
|
+ json={"is_public": False},
|
|
|
+ )
|
|
|
+ assert response.status_code == 401
|
|
|
+
|
|
|
+ # créer un sms simple
|
|
|
+ response = await client.post(
|
|
|
+ app.url_path_for("create_sms_batch", project_id=default_project_id),
|
|
|
+ headers=default_user_headers,
|
|
|
+ json={"template": "Bonjour {prenom}!\n{titre}"},
|
|
|
+ )
|
|
|
+ assert response.status_code == 200
|
|
|
+ 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 == "Bonjour Arthur!\nêtre roi"
|
|
|
+
|
|
|
+
|
|
|
+async def test_delete_project(
|
|
|
+ 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_project", project_id=default_project_id)
|
|
|
+ )
|
|
|
+ assert response.status_code == 401
|
|
|
+ result = session.execute(select(Project).where(Project.id == default_project_id))
|
|
|
+ project = result.scalars().first()
|
|
|
+ assert project is not None
|
|
|
+
|
|
|
+ # Proper deletion
|
|
|
+ response = await client.delete(
|
|
|
+ app.url_path_for("delete_project", project_id=default_project_id),
|
|
|
+ headers=default_user_headers,
|
|
|
+ )
|
|
|
+ assert response.status_code == 200
|
|
|
+ result = session.execute(select(Project).where(Project.id == default_project_id))
|
|
|
+ project = result.scalars().first()
|
|
|
+ assert project is None
|
|
|
+ # check deletion is cascaded to volunteers
|
|
|
+ result = session.execute(
|
|
|
+ select(Volunteer).where(Volunteer.project_id == default_project_id)
|
|
|
+ )
|
|
|
+ volunteer = result.scalars().first()
|
|
|
+ assert volunteer is None
|
|
|
+ # check deletion is cascaded to slots
|
|
|
+ result = session.execute(select(Slot).where(Slot.project_id == default_project_id))
|
|
|
+ slot = result.scalars().first()
|
|
|
+ assert slot is None
|
|
|
+
|
|
|
+ # Idempotence test
|
|
|
+ response = await client.delete(
|
|
|
+ app.url_path_for("delete_project", project_id=default_project_id),
|
|
|
+ headers=default_user_headers,
|
|
|
+ )
|
|
|
+ assert response.status_code == 200
|
|
|
+
|
|
|
+ # can delete random uuid
|
|
|
+ response = await client.delete(
|
|
|
+ app.url_path_for("delete_project", project_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_project", project_id="not uidstr"),
|
|
|
+ headers=default_user_headers,
|
|
|
+ )
|
|
|
+ assert response.status_code == 422
|