Adds views tests for Threads

pyproject.toml

@@ -26,6 +26,7 @@ ipython = "^8.24.0"
 pytest-asyncio = "^0.23.7"
 pytest-recording = "^0.13.1"
 coveralls = "^4.0.1"
+model-bakery = "^1.18.1"
 
 [tool.poetry.group.example.dependencies]
 django-webpack-loader = "^3.1.0"

tests/test_views.py

@@ -1,10 +1,15 @@
+import json
 from http import HTTPStatus
 
+from django.contrib.auth.models import User
+
 import pytest
+from model_bakery import baker
 
 from django_ai_assistant.exceptions import AIAssistantNotDefinedError
 from django_ai_assistant.helpers.assistants import AIAssistant, register_assistant
 from django_ai_assistant.langchain.tools import BaseModel, Field, method_tool
+from django_ai_assistant.models import Thread
 
 
 # Set up
@@ -36,6 +41,13 @@ def fetch_forecast_temperature(self, location: str, dt_str: str) -> str:
         return "35 degrees Celsius"
 
 
+@pytest.fixture
+def authenticated_client(client):
+    User.objects.create_user(username="testuser", password="password")
+    client.login(username="testuser", password="password")
+    return client
+
+
 # Assistant Views
 
 
@@ -70,4 +82,117 @@ def test_does_not_return_assistant_if_unauthorized():
 
 # Threads Views
 
-# Up next
+# GET
+
+
+@pytest.mark.django_db(transaction=True)
+def test_list_threads_without_results(authenticated_client):
+    response = authenticated_client.get("/threads/")
+
+    assert response.status_code == HTTPStatus.OK
+    assert response.json() == []
+
+
+@pytest.mark.django_db(transaction=True)
+def test_list_threads_with_results(authenticated_client):
+    user = User.objects.first()
+    thread = baker.make(Thread, created_by=user)
+    response = authenticated_client.get("/threads/")
+
+    assert response.status_code == HTTPStatus.OK
+    assert response.json()[0].get("id") == thread.id
+
+
+@pytest.mark.django_db(transaction=True)
+def test_does_not_list_other_users_threads(authenticated_client):
+    baker.make(Thread)
+    response = authenticated_client.get("/threads/")
+
+    assert response.status_code == HTTPStatus.OK
+    assert response.json() == []
+
+
+def test_does_not_list_threads_if_unauthorized():
+    # TODO: Implement this test once permissions are in place
+    pass
+
+
+# POST
+
+
+@pytest.mark.django_db(transaction=True)
+def test_create_thread(authenticated_client):
+    response = authenticated_client.post(
+        "/threads/", data=json.dumps({}), content_type="application/json"
+    )
+
+    thread = Thread.objects.first()
+
+    assert response.status_code == HTTPStatus.OK
+    assert response.json().get("id") == thread.id
+
+
+def test_cannot_create_thread_if_unauthorized():
+    # TODO: Implement this test once permissions are in place
+    pass
+
+
+# PATCH
+
+
+@pytest.mark.django_db(transaction=True)
+def test_update_thread(authenticated_client):
+    thread = baker.make(Thread, created_by=User.objects.first())
+    response = authenticated_client.patch(
+        f"/threads/{thread.id}/",
+        data=json.dumps({"name": "New name"}),
+        content_type="application/json",
+    )
+
+    assert response.status_code == HTTPStatus.OK
+
+
+@pytest.mark.django_db(transaction=True)
+def test_cannot_update_other_users_threads(authenticated_client):
+    thread = baker.make(Thread)
+    response = authenticated_client.patch(
+        f"/threads/{thread.id}/",
+        data=json.dumps({"name": "New name"}),
+        content_type="application/json",
+    )
+
+    assert response.status_code == HTTPStatus.FORBIDDEN
+
+
+def test_cannot_update_thread_if_unauthorized():
+    # TODO: Implement this test once permissions are in place
+    pass
+
+
+# DELETE
+
+
+@pytest.mark.django_db(transaction=True)
+def test_delete_thread(authenticated_client):
+    thread = baker.make(Thread, created_by=User.objects.first())
+    response = authenticated_client.delete(f"/threads/{thread.id}/")
+
+    assert response.status_code == HTTPStatus.NO_CONTENT
+
+
+@pytest.mark.django_db(transaction=True)
+def test_cannot_delete_other_users_threads(authenticated_client):
+    thread = baker.make(Thread)
+    response = authenticated_client.delete(f"/threads/{thread.id}/")
+
+    assert response.status_code == HTTPStatus.FORBIDDEN
+
+
+def test_cannot_delete_thread_if_unauthorized():
+    # TODO: Implement this test once permissions are in place
+    pass
+
+
+# Threads Messages Views (will need VCR)
+
+# TBD