Backend Tests for Tasks

.gitignore

@@ -1 +1,2 @@
 .env
+.venv

backend/app/alembic/versions/ba1bdfec52c5_init_initial_database.py

@@ -1,7 +1,7 @@
 """init: initial database
 
 Revision ID: ba1bdfec52c5
-Revises: 
+Revises:
 Create Date: 2024-08-03 22:04:39.316036
 
 """

backend/app/api/routes/tasks.py

@@ -1,10 +1,11 @@
+import uuid
 from typing import Any
 
 from fastapi import APIRouter, HTTPException
 from sqlmodel import func, select
 
 from app.api.deps import CurrentUser, SessionDep
-from app.models import Message, Task, TaskPublic, TasksPublic
+from app.models import Message, Task, TaskCreate, TaskPublic, TasksPublic
 
 router = APIRouter()
 
@@ -40,7 +41,7 @@ def read_tasks(
 
 
 @router.get("/{id}", response_model=TaskPublic)
-def read_task(session: SessionDep, current_user: CurrentUser, id: int) -> Any:
+def read_task(session: SessionDep, current_user: CurrentUser, id: uuid.UUID) -> Any:
     """
     Get task by ID.
     """
@@ -54,7 +55,7 @@ def read_task(session: SessionDep, current_user: CurrentUser, id: int) -> Any:
 
 @router.post("/", response_model=TaskPublic)
 def create_task(
-    *, session: SessionDep, current_user: CurrentUser, task_in: Task
+    *, session: SessionDep, current_user: CurrentUser, task_in: TaskCreate
 ) -> Any:
     """
     Create new task.
@@ -68,7 +69,11 @@ def create_task(
 
 @router.put("/{id}", response_model=TaskPublic)
 def update_task(
-    *, session: SessionDep, current_user: CurrentUser, id: int, task_in: Task
+    *,
+    session: SessionDep,
+    current_user: CurrentUser,
+    id: uuid.UUID,
+    task_in: TaskCreate,
 ) -> Any:
     """
     Update a task.
@@ -87,7 +92,9 @@ def update_task(
 
 
 @router.delete("/{id}")
-def delete_task(session: SessionDep, current_user: CurrentUser, id: int) -> Message:
+def delete_task(
+    session: SessionDep, current_user: CurrentUser, id: uuid.UUID
+) -> Message:
     """
     Delete an task.
     """

backend/app/crud.py

@@ -4,7 +4,7 @@
 from sqlmodel import Session, select
 
 from app.core.security import get_password_hash, verify_password
-from app.models import Item, ItemCreate, User, UserCreate, UserUpdate
+from app.models import Item, ItemCreate, Task, TaskCreate, User, UserCreate, UserUpdate
 
 
 def create_user(*, session: Session, user_create: UserCreate) -> User:
@@ -52,3 +52,11 @@ def create_item(*, session: Session, item_in: ItemCreate, owner_id: uuid.UUID) -
     session.commit()
     session.refresh(db_item)
     return db_item
+
+
+def create_task(*, session: Session, task_in: TaskCreate, owner_id: uuid.UUID) -> Task:
+    db_task = Task.model_validate(task_in, update={"owner_id": owner_id})
+    session.add(db_task)
+    session.commit()
+    session.refresh(db_task)
+    return db_task

backend/app/models.py

@@ -145,6 +145,10 @@ class TaskPublic(TaskBase):
     owner_id: uuid.UUID
 
 
+class TaskCreate(TaskBase):
+    pass
+
+
 class TasksPublic(SQLModel):
     data: list[TaskPublic]
     count: int

backend/app/tests/api/routes/test_items.py

@@ -38,6 +38,8 @@ def test_read_item(
     assert content["description"] == item.description
     assert content["id"] == str(item.id)
     assert content["owner_id"] == str(item.owner_id)
+    assert "id" in content
+    assert "owner_id" in content
 
 
 def test_read_item_not_found(

backend/app/tests/api/routes/test_tasks.py

@@ -0,0 +1,192 @@
+import uuid
+
+from fastapi.testclient import TestClient
+from sqlmodel import Session
+
+from app.core.config import settings
+from app.tests.utils.task import create_random_priority, create_random_task
+
+
+def test_create_task(
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+    db: Session,
+) -> None:
+    priority = create_random_priority(db)
+    data = {
+        "title": "String",
+        "description": "Another String",
+        "priority_id": priority.id,
+        "duration": 0,
+        "due": "2023-10-01T00:00:00",
+    }
+    response = client.post(
+        f"{settings.API_V1_STR}/tasks/",
+        headers=superuser_token_headers,
+        json=data,
+    )
+    assert response.status_code == 200
+    content = response.json()
+    assert content["title"] == data["title"]
+    assert content["description"] == data["description"]
+    assert content["priority_id"] == data["priority_id"]
+    assert content["duration"] == data["duration"]
+    assert "id" in content
+    assert "owner_id" in content
+    # No need to individually asset id and owner id, as we compare to the json, if either doesn't exist it errors
+
+
+def test_read_task(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    task = create_random_task(db)
+    response = client.get(
+        f"{settings.API_V1_STR}/tasks/{task.id}",
+        headers=superuser_token_headers,
+    )
+    assert response.status_code == 200
+    content = response.json()
+    assert content["title"] == task.title
+    assert content["description"] == task.description
+    assert content["priority_id"] == (task.priority_id)
+    assert content["duration"] == (task.duration)
+    assert "id" in content
+    assert "owner_id" in content
+
+
+def test_read_task_not_found(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    response = client.get(
+        f"{settings.API_V1_STR}/tasks/{uuid.uuid4()}",
+        headers=superuser_token_headers,
+    )
+    assert response.status_code == 404
+    content = response.json()
+    assert content["detail"] == "Task not found"
+
+
+def test_read_task_not_enough_permissions(
+    client: TestClient, normal_user_token_headers: dict[str, str], db: Session
+) -> None:
+    task = create_random_task(db)
+    response = client.get(
+        f"{settings.API_V1_STR}/tasks/{task.id}",
+        headers=normal_user_token_headers,
+    )
+    assert response.status_code == 400
+    content = response.json()
+    assert content["detail"] == "Not enough permissions"
+
+
+def test_read_tasks(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    create_random_task(db)
+    create_random_task(db)
+    response = client.get(
+        f"{settings.API_V1_STR}/tasks/",
+        headers=superuser_token_headers,
+    )
+    assert response.status_code == 200
+    content = response.json()
+    assert len(content["data"]) >= 2
+
+
+def test_update_task(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    task = create_random_task(db)
+    priority = create_random_priority(db)
+    data = {
+        "title": "Updated title",
+        "description": "Updated description",
+        "priority_id": priority.id,
+        "duration": 0,
+    }
+    response = client.put(
+        f"{settings.API_V1_STR}/tasks/{task.id}",
+        headers=superuser_token_headers,
+        json=data,
+    )
+    assert response.status_code == 200
+    content = response.json()
+    assert content["title"] == data["title"]
+    assert content["description"] == data["description"]
+    assert content["priority_id"] == data["priority_id"]
+    assert content["duration"] == data["duration"]
+    assert "id" in content
+    assert "owner_id" in content
+
+
+def test_update_task_not_found(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    data = {"title": "Updated title", "description": "Updated description"}
+    response = client.put(
+        f"{settings.API_V1_STR}/tasks/{uuid.uuid4()}",
+        headers=superuser_token_headers,
+        json=data,
+    )
+    assert response.status_code == 404
+    content = response.json()
+    assert content["detail"] == "Task not found"
+
+
+def test_update_item_not_enough_permissions(
+    client: TestClient, normal_user_token_headers: dict[str, str], db: Session
+) -> None:
+    task = create_random_task(db)
+    priority = create_random_priority(db)
+    data = {
+        "title": "Updated title",
+        "description": "Updated description",
+        "priority_id": priority.id,
+        "duration": 100,
+    }
+    response = client.put(
+        f"{settings.API_V1_STR}/tasks/{task.id}",
+        headers=normal_user_token_headers,
+        json=data,
+    )
+    assert response.status_code == 400
+    content = response.json()
+    assert content["detail"] == "Not enough permissions"
+
+
+def test_delete_task(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    task = create_random_task(db)
+    response = client.delete(
+        f"{settings.API_V1_STR}/tasks/{task.id}",
+        headers=superuser_token_headers,
+    )
+    assert response.status_code == 200
+    content = response.json()
+    assert content["message"] == "Task deleted successfully"
+
+
+def test_delete_task_not_found(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    response = client.delete(
+        f"{settings.API_V1_STR}/tasks/{uuid.uuid4()}",
+        headers=superuser_token_headers,
+    )
+    assert response.status_code == 404
+    content = response.json()
+    assert content["detail"] == "Task not found"
+
+
+def test_delete_item_not_enough_permissions(
+    client: TestClient, normal_user_token_headers: dict[str, str], db: Session
+) -> None:
+    task = create_random_task(db)
+    response = client.delete(
+        f"{settings.API_V1_STR}/tasks/{task.id}",
+        headers=normal_user_token_headers,
+    )
+    assert response.status_code == 400
+    content = response.json()
+    assert content["detail"] == "Not enough permissions"

backend/app/tests/utils/task.py

@@ -0,0 +1,50 @@
+import datetime as datetime
+import random
+
+from sqlmodel import Session
+
+from app import crud
+from app.models import Priority, Task, TaskCreate
+from app.tests.utils.user import create_random_user
+from app.tests.utils.utils import (
+    random_lower_string,  # Code from david imports own method, not sure if necessary ?
+)
+
+
+def create_random_priority(db: Session) -> Priority:
+    priorityTitle = random_lower_string()
+    priorityNum = random.randint(1, 9)
+    priority = Priority(name=priorityTitle, value=priorityNum)
+    db.add(priority)
+    db.commit()
+    db.refresh(priority)
+    return priority
+    # based on other code, this should likely be abstracted further to a PriorityCreate model, which creates it's own id (would need to make a new factory for the id too)
+    # Not going to do too much, as I'm not certain this is how it should be implemented. Otherwise it would be done with crud.create_priority or something similar
+    # This abstracts it out from test_tasks.py though
+
+
+def create_random_task(db: Session) -> Task:
+    title = random_lower_string()
+    description = random_lower_string()
+    priority = create_random_priority(db)
+
+    duration = random.randint(1, 100)  # This is just an int, no sure how large it gets
+    year = random.randint(2024, 9999)
+    month = random.randint(1, 12)
+    day = random.randint(
+        1, 29
+    )  # not going higher as random chance of tests failing due to days in months, ie 30th of Feb errors
+    due = datetime.datetime(year, month, day)
+
+    user = create_random_user(db)
+    owner_id = user.id
+    assert owner_id is not None
+    task_in = TaskCreate(
+        title=title,
+        description=description,
+        priority_id=priority.id,
+        duration=duration,
+        due=due,
+    )
+    return crud.create_task(session=db, task_in=task_in, owner_id=owner_id)